const slotName = 'test_slot_wal2json';

const service = new LogicalReplicationService(
  /**
   * node-postgres Client options for connection
   * https://github.com/DefinitelyTyped/DefinitelyTyped/blob/master/types/pg/index.d.ts#L16
   */
  {
    database: 'playground',
    // ...
  },
  /**
   * Logical replication service config
   * https://github.com/kibae/pg-logical-replication/blob/main/src/logical-replication-service.ts#L9
   */
  {
    acknowledge: {
      auto: true,
      timeoutSeconds: 10
    }
  }
)

// `TestDecodingPlugin` for test_decoding and `ProtocolBuffersPlugin` for decoderbufs are also available.
const plugin = new Wal2JsonPlugin({
  /**
   * Plugin options for wal2json
   * https://github.com/kibae/pg-logical-replication/blob/main/src/output-plugins/wal2json/wal2json-plugin-options.type.ts
   */
  //...
});

/**
 * Wal2Json.Output
 * https://github.com/kibae/pg-logical-replication/blob/main/src/output-plugins/wal2json/wal2json-plugin-output.type.ts
 */
service.on('data', (lsn: string, log: Wal2Json.Output) => {
  // Do something what you want.
  // log.change.filter((change) => change.kind === 'insert').length;
});

// Start subscribing to data change events.
(function proc() {
  service.subscribe(plugin, slotName)
    .catch((e) => {
      console.error(e);
    })
    .then(() => {
      setTimeout(proc, 100);
    });
})();

const service = new LogicalReplicationService({
  /**
   * node-postgres Client options for connection
   * https://github.com/DefinitelyTyped/DefinitelyTyped/blob/master/types/pg/index.d.ts#L16
   */
  clientConfig: {
    user? : string | undefined;
    database? : string | undefined;
    password? : string | (() => string | Promise<string>) | undefined;
    port? : number | undefined;
    host? : string | undefined;
    connectionString? : string | undefined;
    keepAlive? : boolean | undefined;
    stream? : stream.Duplex | undefined;
    statement_timeout? : false | number | undefined;
    parseInputDatesAsUTC? : boolean | undefined;
    ssl? : boolean | ConnectionOptions | undefined;
    query_timeout? : number | undefined;
    keepAliveInitialDelayMillis? : number | undefined;
    idle_in_transaction_session_timeout? : number | undefined;
    application_name? : string | undefined;
    connectionTimeoutMillis? : number | undefined;
    types? : CustomTypesConfig | undefined;
    options? : string | undefined;
  },
  /**
   * Logical replication service config
   * https://github.com/kibae/pg-logical-replication/blob/main/src/logical-replication-service.ts#L9
   */
  config? : Partial<{
    acknowledge?: {
      /**
       * If the value is false, acknowledge must be done manually.
       * Default: true
       */
      auto: boolean;
      /**
       * Acknowledge is performed every set time (sec). If 0, do not do it.
       * Default: 10
       */
      /**
       * Periodically send standby status (heartbeat) to PostgreSQL and acknowledge WAL progress.
       * Only takes effect when auto is true. If 0, do not send periodic acknowledgements.
       * Default: 10
       */
      timeoutSeconds: 0 | 10 | number;
    };
    flowControl?: {
      /**
       * If true, pause the stream until the data handler completes.
       * This enables backpressure support for async handlers.
       * Default: false
       */
      enabled: boolean;
    };
  }>
})

const service = new LogicalReplicationService(clientConfig, {
  acknowledge: {
    auto: false,      // Disable automatic acknowledgement
    timeoutSeconds: 0 // Disable periodic standby status as well
  }
});

service.on('data', async (lsn: string, log: Wal2Json.Output) => {
  try {
    // Process the change
    await processChange(log);

    // Manually acknowledge only after successful processing.
    // PostgreSQL will not advance the replication slot until this is called.
    await service.acknowledge(lsn);
  } catch (err) {
    // If you don't acknowledge, PostgreSQL will re-send this change on reconnect.
    console.error('Failed to process, skipping ack:', err);
  }
});

const service = new LogicalReplicationService(clientConfig, {
  acknowledge: { auto: true, timeoutSeconds: 10 },
  flowControl: { enabled: true }  // Enable backpressure support
});

// Now async handlers are fully supported - the stream pauses until processing completes
service.on('data', async (lsn: string, log: Pgoutput.Message) => {
  await someSlowAsyncOperation(log);  // Safe: next message waits for this to complete
});