@spec count_matching(keyword(), keyword()) :: {:ok, %{count: non_neg_integer()}}

Counts changes matching filters without loading row payloads.

Same validation and join semantics as Threadline.Query.timeline/2.

Options

• :repo — optional if :repo is present in filters
• :cap — when set to a positive integer, the count short-circuits at that value via a windowed subquery (SELECT count(*) FROM (... LIMIT ^cap)), so multi-million-row tables return immediately at the cap rather than waiting for a full aggregate scan. The default (nil) preserves the existing unbounded behavior. The Mix task mix threadline.export does NOT pass :cap and is unaffected; Threadline.OperatorSurface.Live.TimelineLive and the export controller pass cap: 10_001 so the LV can render "10,000+ matches" without hitting statement_timeout.

@spec csv_header(keyword()) :: iodata()

Returns the canonical CSV header row (one line ending in \r\n) for the operator-surface export controller's chunked path.

Same column order as to_csv_iodata/2. The chunked path emits this as the first chunk before streaming data rows so the byte-equality parity test holds against the iodata path.

Options

• :include_action_metadata — when true, append correlation_id and action_id columns (same shape as to_csv_iodata/2).

@spec format_changes_iodata([struct()], :csv | :json_wrapped | :ndjson, keyword()) ::
  iodata()

Formats a pre-fetched list of %AuditChange{} structs into iodata for the requested format.

Used by the operator-surface export controller to format streamed batches; CSV header and JSON envelopes are NOT emitted by this function and remain the caller's responsibility (the chunked path emits the header / envelope as its first / last chunk).

• :csv — CSV data rows (each terminated by \r\n per RFC 4180); the caller MUST emit csv_header/1 as the first chunk.
• :json_wrapped — each row as Jason.encode!/1 output (a JSON object). The caller emits the surrounding {"format_version": ..., "generated_at": ..., "changes": [ prefix and ]} suffix as separate chunks plus the inter-row comma separators.
• :ndjson — each row as Jason.encode!/1 output followed by \n (no envelope; pure line-delimited JSON).

Options

• :include_action_metadata (default false) — same shape as to_csv_iodata/2.

@spec stream_changes(keyword(), keyword()) :: Enumerable.t()

Lazily enumerates AuditChange structs in timeline order using keyset pages.

Does not enforce max_rows — combine with Stream.take/2 if needed.

Options

• :repo — optional if present in filters
• :page_size — defaults to 1000

@spec stream_export_rows(keyword(), keyword()) :: Enumerable.t()

Lazily enumerates the join-projected export-row maps (same shape as to_csv_iodata/2 / to_json_document/2 consume) in timeline order using keyset pages.

Each emitted item is a map with the keys :id, :transaction_id, :table_schema, :table_name, :op, :captured_at, :table_pk, :data_after, :changed_fields, :changed_from, :tx_occurred_at, :tx_actor_ref, :tx_source, :aa_id, :aa_correlation_id — exactly the projection from Threadline.Query.export_changes_query/1. This matches what format_changes_iodata/3 expects, so the operator-surface export controller's chunked path produces byte-identical output to the iodata path.

Does not enforce max_rows — combine with Stream.take/2 if needed.

Options

• :repo — optional if present in filters
• :page_size — defaults to 1000

@spec to_csv_iodata(keyword(), keyword()) :: {:ok, map()}

Returns CSV as iodata plus truncation metadata.

See module documentation for filters, opts, and column layout.

Options

• :repo — optional if :repo is present in filters
• :max_rows — defaults to 10000
• :include_action_metadata — when true, append correlation_id and action_id columns

@spec to_json_document(keyword(), keyword()) :: {:ok, map()}

Returns JSON (wrapped object or NDJSON lines) as iodata plus truncation metadata.

Options

• :repo, :max_rows — same as to_csv_iodata/2
• :json_format — :wrapped (default) or :ndjson