How to annotate database changes

select pg_logical_emit_message(true, 'sequin:transaction_annotations.set', '{
  "username": "paul.atreides",
  "source": "web_ui",
  "request_id": "req_123"
}');

{
  "record": { /* ... */ },
  "changes": null,
  "action": "insert",
  "metadata": {
    "table_schema": "public",
    "table_name": "orders",
    "commit_timestamp": "2024-10-28T21:37:53Z",
    "commit_lsn": 123456789,
    "commit_idx": 1,
    "database_name": "myapp-prod",
    "consumer": {
      "id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
      "name": "orders_kafka",
      "annotations": {"my-custom-key": "my-custom-value"}
    },
    "database": {
      "id": "12345678-9abc-def0-1234-56789abcdef0",
      "name": "myapp-prod",
      "annotations": {"my-custom-key": "my-custom-value"},
      "database": "myapp-prod",
      "hostname": "db.example.com"
    },
    "transaction_annotations": {
      "username": "paul.atreides",
      "source": "web_ui",
      "request_id": "req_123"
    }
  }
}

begin;
  -- Set annotations for this transaction
  select pg_logical_emit_message(true, 'sequin:transaction_annotations.set', '{
    "username": "paul.atreides",
    "source": "web_ui",
    "request_id": "req_123"
  }');

  -- All changes in this transaction will include the annotations
  insert into orders (customer_id, product, quantity)
  values (789, 'Wireless Headphones', 1);

  update customers
  set last_order_at = now()
  where id = 789;
commit;

select pg_logical_emit_message(true, 'sequin:transaction_annotations.clear', '');

begin;
  select pg_logical_emit_message(true, 'sequin:transaction_annotations.set', '{
    "username": "paul.atreides",
    "action_type": "refund_order",
    "reason": "customer_request",
    "ticket_id": "TICKET-123"
  }');

  update orders set status = 'refunded' where id = 456;
  insert into refunds (order_id, amount) values (456, 99.99);
commit;

begin;
  select pg_logical_emit_message(true, 'sequin:transaction_annotations.set', '{
    "request_id": "req_123",
    "trace_id": "trace_456",
    "user_agent": "Mozilla/5.0...",
    "ip_address": "1.2.3.4"
  }');

  -- Make changes
commit;

begin;
  select pg_logical_emit_message(true, 'sequin:transaction_annotations.set', '{
    "workflow": "new_subscription",
    "plan_id": "pro_monthly",
    "payment_id": "py_789",
    "coupon_code": "WELCOME"
  }');

  insert into subscriptions (...);
  insert into subscription_items (...);
  update customer_metrics (...);
commit;