Matcher events

Matcher has 46 catalog definitions and production emission paths for 39 of them. It emits domain events as CloudEvents 1.0 messages in binary content mode over Kafka, published through lib-streaming. Every event travels in the shared envelope: ce-type names the event as studio.lerian.matcher.<resource>.<event>, ce-subject carries the aggregate id, ce-tenantid the owning tenant, and ce-schemaversion the payload version. The version is 1.0.0 for every event below. ce-source comes from STREAMING_CLOUDEVENTS_SOURCE, which every deployment with streaming enabled must set. Deployments use matcher. With that value, every event uses the application topic lerian.streaming.matcher and the source-qualified types shown below. A different configured source changes both. Resource type, event type, and schema version remain in CloudEvents headers. Money amounts (fee values and adjustment amounts) cross the wire as decimal strings, never floats. Matcher serves its full event catalog at GET /system/matcher/streaming/manifest. This page covers the Kafka streaming plane. Matcher’s exception webhook dispatch (HTTP callbacks for exception routing) is a separate surface, documented in Webhooks and callbacks.

Delivery policies

Matcher’s catalog uses two delivery policies: 16 CRITICAL definitions use the outbox-backed policy and 30 IMPORTANT definitions use direct delivery with circuit-open fallback.

Outbox-backed events go to the outbox in the same database transaction as the state change. A relay publishes committed rows and retries through broker outages. These are the audit-grade facts (matching outcomes, exception resolutions, disputes, governance). A deployment cannot weaken this policy.
Direct events publish after the transaction commits, best-effort, falling back to the outbox only when the broker circuit is open. These are configuration and operational lifecycle signals.

Each table below states the policy for its events.

Matching events

Outbox-backed: transaction.matched, transaction.pending_review. Direct: the rest.

Event (`ce-type`)	Topic	Fires when	Key payload
`studio.lerian.matcher.transaction.matched`	`lerian.streaming.matcher`	A transaction line reaches terminal MATCHED in a match-run commit — one event per matched line.	`transaction_id`, `context_id`, `match_run_id`, `previous_status`, `status`, `match_group_id`?, `candidate_match_group_id`?, `source_id`?, `matched_at`
`studio.lerian.matcher.transaction.pending_review`	`lerian.streaming.matcher`	A candidate match needs human review (non-automatic rule).	Same schema, with required `pending_review_at`
`studio.lerian.matcher.transaction.ignored`	`lerian.streaming.matcher`	A transaction is excluded from matching.	`transaction_id`, `ingestion_job_id`, `context_id`, `source_id`, `previous_status`, `status`, `extraction_status`, `updated_at`
`studio.lerian.matcher.match_run.completed`	`lerian.streaming.matcher`	A match run finishes. Lender consumes this event to translate reconciliation verdicts into loan servicing actions.	`match_run_id`, `context_id`, `mode`, `status`, `stats`, `started_at`, `completed_at`?
`studio.lerian.matcher.match_run.failed`	`lerian.streaming.matcher`	A match run fails.	`match_run_id`, `context_id`, `mode`, `status`, `failure_reason`, `started_at`, `completed_at`?
`studio.lerian.matcher.match_group.confirmed`	`lerian.streaming.matcher`	A match group is confirmed.	`match_group_id`, `match_run_id`, `context_id`, `rule_id`, `transaction_ids`, `confidence`, `status`, `confirmed_at`?
`studio.lerian.matcher.match_group.unmatched`	`lerian.streaming.matcher`	A confirmed match group is undone.	`match_group_id`, `match_run_id`, `context_id`, `rule_id`, `transaction_ids`, `previous_status`, `status`, `reason`, `unmatched_at`
`studio.lerian.matcher.fee_variance.created`	`lerian.streaming.matcher`	A fee-aware rule detects a variance between expected and actual fees.	`fee_variance_id`, `context_id`, `match_run_id`, `match_group_id`, `transaction_id`, `fee_schedule_id`, `fee_schedule_name_snapshot`, `currency`, `expected_fee`, `actual_fee`, `delta`, `variance_type`, `created_at`

Exception and dispute events

Outbox-backed except exception.assigned and the comment events, which are direct.

Event (`ce-type`)	Topic	Fires when	Key payload
`studio.lerian.matcher.exception.assigned`	`lerian.streaming.matcher`	An exception is assigned to an operator.	`exception_id`, `status`, `version`, `resolution_type`?, `transaction_id`?, `assigned_at`
`studio.lerian.matcher.exception.resolved`	`lerian.streaming.matcher`	An exception is resolved.	`exception_id`, `status`, `version`, `resolution_type`?, `transaction_id`?, `resolved_at`
`studio.lerian.matcher.exception.force_match_resolved`	`lerian.streaming.matcher`	An exception is resolved by force-match with an override reason.	Base fields plus `resolution_type` (`FORCE_MATCH`), `override_reason_code`, `resolved_at`
`studio.lerian.matcher.exception.adjust_entry_resolved`	`lerian.streaming.matcher`	An exception is resolved by an adjustment entry.	Base fields plus `resolution_type` (`ADJUST_ENTRY`), `reason_code`, `amount`, `currency`, `resolved_at`
`studio.lerian.matcher.exception.dispatched`	`lerian.streaming.matcher`	An exception is dispatched to an external system.	`exception_id`, `target_system`, `queue`, `external_reference`, `acknowledged`, `dispatched_at`
`studio.lerian.matcher.exception.callback_processed`	`lerian.streaming.matcher`	An external system’s callback is applied to an exception.	Base fields plus `external_system`, `external_issue_id`, `callback_type`, `processed_at`
`studio.lerian.matcher.exception_comment.added`	`lerian.streaming.matcher`	A comment is added to an exception thread.	`comment_id`, `exception_id`, `created_at`
`studio.lerian.matcher.exception_comment.deleted`	`lerian.streaming.matcher`	A comment is deleted.	`comment_id`, `exception_id`, `actor`, `deleted_at`
`studio.lerian.matcher.dispute.opened`	`lerian.streaming.matcher`	A dispute is opened on an exception.	`dispute_id`, `exception_id`, `state`, `resolution`?, `category`, `opened_at`
`studio.lerian.matcher.dispute.won`	`lerian.streaming.matcher`	A dispute closes as won.	`dispute_id`, `exception_id`, `state`, `resolution`?, `closed_at`
`studio.lerian.matcher.dispute.lost`	`lerian.streaming.matcher`	A dispute closes as lost.	`dispute_id`, `exception_id`, `state`, `resolution`?, `closed_at`
`studio.lerian.matcher.evidence.submitted`	`lerian.streaming.matcher`	Evidence is attached to a dispute.	`evidence_id`, `dispute_id`, `exception_id`, `has_file`, `submitted_at`

Configuration and data-intake events

All direct.

Event (`ce-type`)	Topic	Fires when	Key payload
`studio.lerian.matcher.reconciliation_context.created`	`lerian.streaming.matcher`	A reconciliation context is created.	`context_id`, `name`, `context_type`, `interval`, `status`, `auto_match_on_upload`, `created_at`, `tenant_id`
`studio.lerian.matcher.reconciliation_context.updated`	`lerian.streaming.matcher`	A context’s metadata or lifecycle status changes.	`context_id`, `name`, `context_type`, `interval`, `status`, `auto_match_on_upload`, `updated_at`, `tenant_id`
`studio.lerian.matcher.reconciliation_source.created`	`lerian.streaming.matcher`	A source (input side) is created in a context.	`context_id`, `source_id`, `name`, `source_type`, `side`, `created_at`, `tenant_id`
`studio.lerian.matcher.match_rule.created`	`lerian.streaming.matcher`	A match rule is created.	`context_id`, `rule_id`, `rule_type`, `priority`, `config_hash`, `created_at`, `tenant_id`
`studio.lerian.matcher.match_rule.reordered`	`lerian.streaming.matcher`	Rule priorities are reordered.	`context_id`, `ordered_rule_ids`, `priority_version`, `reordered_at`, `tenant_id`
`studio.lerian.matcher.fetcher_connection.synced`	`lerian.streaming.matcher`	A Discovery connection and its discovered schema snapshot are synchronized.	`connection_id`, `fetcher_connection_id`, `config_name`, `database_type`, `status`, `schema_discovered`, `last_seen_at`, `updated_at`
`studio.lerian.matcher.fetcher_connection.unreachable`	`lerian.streaming.matcher`	A Discovery connection becomes unreachable.	`connection_id`, `fetcher_connection_id`, `previous_status`, `status`, `schema_discovered`, `updated_at`
`studio.lerian.matcher.extraction_request.created`	`lerian.streaming.matcher`	An extraction request is created against a connection.	`extraction_request_id`, `connection_id`, `status`, `table_count`, `has_filters`, `start_date`, `end_date`, `created_at`
`studio.lerian.matcher.ingestion.completed`	`lerian.streaming.matcher`	An ingestion job completes.	`job_id`, `context_id`, `source_id`, `status`, `total_rows`, `failed_rows`, `transaction_count`, `date_range_start`?, `date_range_end`?, `completed_at`?, `error_code`?
`studio.lerian.matcher.ingestion.failed`	`lerian.streaming.matcher`	An ingestion job fails.	`job_id`, `context_id`, `source_id`, `status`, `total_rows`, `failed_rows`, `error_code`?, `failed_at`?

Governance and reporting events

Outbox-backed except the export-job family, which is direct.

Event (`ce-type`)	Topic	Fires when	Key payload
`studio.lerian.matcher.audit_log.created`	`lerian.streaming.matcher`	A hash-chained audit log entry is persisted.	`audit_log_id`, `tenant_id`, `entity_type`, `entity_id`, `action`, `tenant_seq`, `hash_version`, `record_hash`, `created_at`
`studio.lerian.matcher.archive_metadata.created`	`lerian.streaming.matcher`	The archival worker registers an archive partition.	`archive_metadata_id`, `tenant_id`, `partition_name`, `date_range_start`, `date_range_end`, `status`, `created_at`, `updated_at`, plus `checksum`?, `row_count`?, `compressed_size_bytes`?, `storage_class`?, `archived_at`?
`studio.lerian.matcher.archive.uploaded`	`lerian.streaming.matcher`	An archive is uploaded to storage.	Same schema
`studio.lerian.matcher.archive.completed`	`lerian.streaming.matcher`	An archival cycle completes.	Same schema
`studio.lerian.matcher.actor.pseudonymized`	`lerian.streaming.matcher`	An actor’s personal data is pseudonymized (GDPR/LGPD).	`actor_id`, `pseudonymized`, `display_name_status`, `email_status`, `updated_at`, `tenant_id`
`studio.lerian.matcher.export_job.created`	`lerian.streaming.matcher`	An export job is created.	`export_job_id`, `tenant_id`, `context_id`, `report_type`, `format`, `status`, `schema_version`, `created_at`, `expires_at`, `updated_at`
`studio.lerian.matcher.export_job.succeeded`	`lerian.streaming.matcher`	An export job finishes writing its artifact.	Same schema, plus `file_name`?, `sha256`?, `records_written`?, `bytes_written`?, `attempts`?, `finished_at`?
`studio.lerian.matcher.export_job.failed`	`lerian.streaming.matcher`	An export job fails.	Same schema, plus `error_code`, `attempts`?, `finished_at`?
`studio.lerian.matcher.export_job.expired`	`lerian.streaming.matcher`	An export artifact expires and is cleaned up.	Same schema, plus `expired_at`?

Payload contracts

These types mirror the CloudEvent data objects emitted at matcher@e3902325. ? means the property can be absent. Timestamps are UTC RFC 3339 with nanosecond precision. UUID aliases document semantic identity. A field typed as string carries no UUID guarantee. stats is an open integer map, so consumers must tolerate new keys.

type RFC3339Nano = string
type DecimalString = string
type UUID = string
type MatchRunStats = Record<string, number>

// reconciliation_context.created
type ReconciliationContextCreatedDataV1 = {
  context_id: UUID
  name: string
  context_type: string
  interval: string
  status: string
  auto_match_on_upload: boolean
  created_at: RFC3339Nano
  tenant_id: string
}

// reconciliation_context.updated
type ReconciliationContextUpdatedDataV1 = {
  context_id: UUID
  name: string
  context_type: string
  interval: string
  status: string
  auto_match_on_upload: boolean
  updated_at: RFC3339Nano
  tenant_id: string
}

// reconciliation_source.created
type ReconciliationSourceCreatedDataV1 = {
  context_id: UUID
  source_id: UUID
  name: string
  source_type: string
  side: string
  created_at: RFC3339Nano
  tenant_id: string
}

// match_rule.created
type MatchRuleCreatedDataV1 = {
  context_id: UUID
  rule_id: UUID
  rule_type: string
  priority: number
  config_hash: string
  created_at: RFC3339Nano
  tenant_id: string
}

// match_rule.reordered
type MatchRuleReorderedDataV1 = {
  context_id: UUID
  ordered_rule_ids: UUID[]
  priority_version: string
  reordered_at: RFC3339Nano
  tenant_id: string
}

// fetcher_connection.synced
type FetcherConnectionSyncedDataV1 = {
  connection_id: UUID
  fetcher_connection_id: string
  config_name: string
  database_type: string
  status: string
  schema_discovered: boolean
  last_seen_at: RFC3339Nano
  updated_at: RFC3339Nano
}

// fetcher_connection.unreachable
type FetcherConnectionUnreachableDataV1 = {
  connection_id: UUID
  fetcher_connection_id: string
  previous_status: string
  status: string
  schema_discovered: boolean
  updated_at: RFC3339Nano
}

// extraction_request.created
type ExtractionRequestCreatedDataV1 = {
  extraction_request_id: UUID
  connection_id: UUID
  status: string
  table_count: number
  has_filters: boolean
  start_date: string
  end_date: string
  created_at: RFC3339Nano
}

// ingestion.completed
type IngestionCompletedDataV1 = {
  job_id: UUID
  context_id: UUID
  source_id: UUID
  status: string
  total_rows: number
  failed_rows: number
  transaction_count: number
  date_range_start?: RFC3339Nano
  date_range_end?: RFC3339Nano
  completed_at?: RFC3339Nano
  error_code?: string
}

// ingestion.failed
type IngestionFailedDataV1 = {
  job_id: UUID
  context_id: UUID
  source_id: UUID
  status: string
  total_rows: number
  failed_rows: number
  failed_at?: RFC3339Nano
  error_code?: string
}

// transaction.ignored
type TransactionIgnoredDataV1 = {
  transaction_id: UUID
  ingestion_job_id: UUID
  context_id: UUID
  source_id: UUID
  previous_status: string
  status: string
  extraction_status: string
  updated_at: RFC3339Nano
}

// transaction.matched
type TransactionMatchedDataV1 = {
  transaction_id: UUID
  context_id: UUID
  match_run_id: UUID
  previous_status: string
  status: string
  match_group_id?: UUID
  candidate_match_group_id?: UUID
  source_id?: UUID
  matched_at: RFC3339Nano
}

// transaction.pending_review
type TransactionPendingReviewDataV1 = {
  transaction_id: UUID
  context_id: UUID
  match_run_id: UUID
  previous_status: string
  status: string
  match_group_id?: UUID
  candidate_match_group_id?: UUID
  source_id?: UUID
  pending_review_at: RFC3339Nano
}

// match_run.completed
type MatchRunCompletedDataV1 = {
  match_run_id: UUID
  context_id: UUID
  mode: string
  status: string
  stats: MatchRunStats
  started_at: RFC3339Nano
  completed_at?: RFC3339Nano
}

// match_run.failed
type MatchRunFailedDataV1 = {
  match_run_id: UUID
  context_id: UUID
  mode: string
  status: string
  failure_reason: string
  started_at: RFC3339Nano
  completed_at?: RFC3339Nano
}

// match_group.confirmed
type MatchGroupConfirmedDataV1 = {
  match_group_id: UUID
  match_run_id: UUID
  context_id: UUID
  rule_id: UUID
  transaction_ids: UUID[]
  confidence: number
  status: string
  confirmed_at?: RFC3339Nano
}

// match_group.unmatched
type MatchGroupUnmatchedDataV1 = {
  match_group_id: UUID
  match_run_id: UUID
  context_id: UUID
  rule_id: UUID
  transaction_ids: UUID[]
  previous_status: string
  status: string
  reason: string
  unmatched_at: RFC3339Nano
}

// fee_variance.created
type FeeVarianceCreatedDataV1 = {
  fee_variance_id: UUID
  context_id: UUID
  match_run_id: UUID
  match_group_id: UUID
  transaction_id: UUID
  fee_schedule_id: UUID
  fee_schedule_name_snapshot: string
  currency: string
  expected_fee: DecimalString
  actual_fee: DecimalString
  delta: DecimalString
  variance_type: string
  created_at: RFC3339Nano
}

type ExceptionDataBaseV1 = {
  exception_id: UUID
  status: string
  version: number
  resolution_type?: string
  transaction_id?: UUID
}

// exception.assigned
type ExceptionAssignedDataV1 = ExceptionDataBaseV1 & {
  assigned_at: RFC3339Nano
}

// exception.resolved
type ExceptionResolvedDataV1 = ExceptionDataBaseV1 & {
  resolved_at: RFC3339Nano
}

// exception.callback_processed
type ExceptionCallbackProcessedDataV1 = ExceptionDataBaseV1 & {
  external_system: string
  external_issue_id: string
  callback_type: string
  processed_at: RFC3339Nano
}

// exception.force_match_resolved
type ExceptionForceMatchResolvedDataV1 = ExceptionDataBaseV1 & {
  resolution_type: "FORCE_MATCH"
  override_reason_code: string
  resolved_at: RFC3339Nano
}

// exception.adjust_entry_resolved
type ExceptionAdjustEntryResolvedDataV1 = ExceptionDataBaseV1 & {
  resolution_type: "ADJUST_ENTRY"
  reason_code: string
  amount: DecimalString
  currency: string
  resolved_at: RFC3339Nano
}

// exception.dispatched
type ExceptionDispatchedDataV1 = {
  exception_id: UUID
  target_system: string
  queue: string
  external_reference: string
  acknowledged: boolean
  dispatched_at: RFC3339Nano
}

// exception_comment.added
type ExceptionCommentAddedDataV1 = {
  comment_id: UUID
  exception_id: UUID
  created_at: RFC3339Nano
}

// exception_comment.deleted
type ExceptionCommentDeletedDataV1 = {
  comment_id: UUID
  exception_id: UUID
  actor: string
  deleted_at: RFC3339Nano
}

// dispute.opened
type DisputeOpenedDataV1 = {
  dispute_id: UUID
  exception_id: UUID
  state: string
  resolution?: string
  category: string
  opened_at: RFC3339Nano
}

// dispute.won and dispute.lost
type DisputeClosedDataV1 = {
  dispute_id: UUID
  exception_id: UUID
  state: string
  resolution?: string
  closed_at: RFC3339Nano
}

// evidence.submitted
type EvidenceSubmittedDataV1 = {
  evidence_id: UUID
  dispute_id: UUID
  exception_id: UUID
  has_file: boolean
  submitted_at: RFC3339Nano
}

// audit_log.created
type AuditLogCreatedDataV1 = {
  audit_log_id: UUID
  tenant_id: UUID
  entity_type: string
  entity_id: UUID
  action: string
  tenant_seq: number
  hash_version: number
  record_hash: string
  created_at: RFC3339Nano
}

// archive_metadata.created, archive.uploaded, and archive.completed
type ArchiveDataV1 = {
  archive_metadata_id: UUID
  tenant_id: UUID
  partition_name: string
  date_range_start: RFC3339Nano
  date_range_end: RFC3339Nano
  status: string
  created_at: RFC3339Nano
  updated_at: RFC3339Nano
  checksum?: string
  row_count?: number
  compressed_size_bytes?: number
  storage_class?: string
  archived_at?: RFC3339Nano
}

// actor.pseudonymized
type ActorPseudonymizedDataV1 = {
  actor_id: string
  pseudonymized: boolean
  display_name_status: "REDACTED"
  email_status: "REDACTED"
  updated_at: RFC3339Nano
  tenant_id: string
}

// export_job.created, export_job.succeeded, export_job.failed, and export_job.expired
type ExportJobDataV1 = {
  export_job_id: UUID
  tenant_id: UUID
  context_id: UUID
  report_type: string
  format: string
  status: string
  schema_version: "1.0.0"
  created_at: RFC3339Nano
  expires_at: RFC3339Nano
  updated_at: RFC3339Nano
  file_name?: string
  sha256?: string
  records_written?: number
  bytes_written?: number
  attempts?: number
  error_code?: "EXPORT_JOB_FAILED"
  finished_at?: RFC3339Nano
  expired_at?: RFC3339Nano
}

ingestion.completed includes the date range only when the pipeline calculated one. Terminal timestamps and stable error codes depend on the persisted aggregate. Export counters are absent while zero. dispute.won and dispute.lost do not repeat the category sent by dispute.opened.

Declared but not yet emitted

The catalog and manifest declare 46 definitions, but production code emits 39. These seven definitions have no non-test caller:

Definition key	Current status
`reconciliation_context.deleted`	Contract reservation; no production emission path.
`extraction_request.submitted`	Contract reservation; state transitions do not emit it.
`extraction_request.completed`	Contract reservation; state transitions do not emit it.
`extraction_request.failed`	Contract reservation; state transitions do not emit it.
`extraction_request.cancelled`	Contract reservation; state transitions do not emit it.
`extraction_request.bridged`	Contract reservation; state transitions do not emit it.
`extraction_request.bridge_failed`	Contract reservation; state transitions do not emit it.

Do not build consumers that depend on receiving these definitions until Matcher adds and wires production emitters.

Events consumed

Matcher consumes no Kafka streaming events. Its integration inputs arrive over HTTP: file uploads, Discovery extractions, and, when enabled, Streaming Hub webhook deliveries. Those deliveries land on POST /v1/discovery/hub/events (HMAC-signed, deduplicated by event id, disabled by default).

​Delivery policies

​Matching events

​Exception and dispute events

​Configuration and data-intake events

​Governance and reporting events

​Payload contracts

​Declared but not yet emitted

​Events consumed