audit

func (*ConsentStore) HasActiveGrant ¶

func (c *ConsentStore) HasActiveGrant(emailHash string) (bool, error)

HasActiveGrant reports whether the email-hash has at least one grant row that has not been subsequently withdrawn. Returns (true, nil) when an active consent exists; (false, nil) when none does (either never granted, or every grant has been withdrawn).

Used by tooling that must verify consent before performing user- affecting work — e.g. scheduled briefings, marketing dispatch.

func (*ConsentStore) InitTable ¶

func (c *ConsentStore) InitTable() error

InitTable creates the consent_log table and its supporting indexes if they do not already exist. Idempotent — safe to call at every startup.

The CHECK constraints on consent_action and method mirror the ConsentAction and ConsentMethod constants. Extending those constants requires an ALTER-TABLE-recreate migration (SQLite can't drop CHECK constraints).

Migration sequencing matters: SQLite ADD COLUMN must complete before any partial-index references the column on an existing table. Earlier versions (pre-PR-D) of this DDL ran the partial index `idx_consent_active` (which references `withdrawn_at IS NULL`) BEFORE the ALTER TABLE migration, so on pre-PR-D databases the index creation failed before the column could be added — taking down the whole startup. We now split into three phases:

1. CREATE TABLE — idempotent, no-op on existing tables.
2. ALTER TABLE ADD COLUMN — idempotent via duplicate-column tolerance.
3. CREATE INDEX — now safe; the column is guaranteed to exist.

func (*ConsentStore) Insert ¶

func (c *ConsentStore) Insert(entry *ConsentLogEntry) error

Insert persists a consent event. On success, entry.ID is populated with the auto-increment row ID.

Timestamp handling: we store the caller-supplied TimestampUTC explicitly rather than relying on CURRENT_TIMESTAMP, so that replayed events (e.g. a background worker draining a queue after restart) carry their original timestamp. Caller should always pass time.Now().UTC().

func (*ConsentStore) ListByEmailHash ¶

func (c *ConsentStore) ListByEmailHash(emailHash string, limit int) ([]*ConsentLogEntry, error)

ListByEmailHash returns all consent events for the given user_email_hash ordered by timestamp_utc ASC (oldest first, so the consent history reads chronologically). Limit caps the result set; 0 or negative means "no cap".

func (*ConsentStore) MarkWithdrawnByEmailHash ¶

func (c *ConsentStore) MarkWithdrawnByEmailHash(
	emailHash string,
	withdrawnAt time.Time,
	noticeVersion, reason, ipAddress, userAgent string,
) (int64, error)

MarkWithdrawnByEmailHash stamps every active grant row for the given user_email_hash with withdrawnAt and inserts a "withdraw" action row describing the withdrawal. Both writes happen in sequence (SQLite single-writer pool serialises them); a partial failure leaves the log internally consistent because grant rows are always findable via consent_action='grant' AND withdrawn_at IS NULL.

noticeVersion identifies which privacy notice the user accepted at the time their grant was captured — copied onto the withdraw row so the audit trail records "user withdrew consent to <version> at <timestamp>". Caller passes the active notice version; ListByEmail Hash can recover the original grant's notice from its own row.

reason is plain text recorded as the withdraw row's scope field (DPDP §6(4) doesn't require a reason; we capture it for operations).

Returns the number of grant rows that were marked withdrawn. Zero is a normal outcome when the user had nothing to withdraw — the caller can decide whether to surface that as an error.

func (HashPublishConfig) Enabled ¶

func (c HashPublishConfig) Enabled() bool

Enabled reports whether the publisher has enough configuration to run. A publisher with Enabled()==false logs a single "disabled" line at startup and becomes a no-op. This keeps local dev and unconfigured deployments working without surprise failures.

func (*Store) AddActivityListener ¶

func (s *Store) AddActivityListener(id string) chan *ToolCall

AddActivityListener registers a buffered channel that receives new ToolCall entries as they are recorded.

func (*Store) ChainTip ¶

func (s *Store) ChainTip() (string, int64, error)

ChainTip returns the latest entry_hash and the total count of audit entries. Used by the hash publisher. (Chain-break markers are included in the count because they are legitimate chain links.)

Returns ("", 0, nil) on an empty chain — callers should treat this as "nothing to publish yet", not an error.

func (*Store) CleanupOldRecords ¶

func (s *Store) CleanupOldRecords(days int) (int64, error)

CleanupOldRecords deletes tool_call rows older than `days` days from now. Returns the number of rows deleted.

When days <= 0 retention is treated as DISABLED and the call is a no-op returning (0, nil). Operators disable the in-package cleanup worker by setting AUDIT_RETENTION_DAYS=0 (useful when an external scheduler already owns retention, e.g. app/wire.go).

Internally delegates to DeleteOlderThan so the hash-chain break-marker behaviour applies: chain integrity is preserved across retention deletions via a recorded __chain_break entry. See VerifyChain for details.

func (*Store) DeleteOlderThan ¶

func (s *Store) DeleteOlderThan(before time.Time) (int64, error)

DeleteOlderThan removes tool_calls older than the given time. If hash chaining is active, a chain-break marker is inserted before the deletion to preserve chain continuity for verification. Returns the number of rows deleted (excluding the marker).

func (*Store) DroppedCount ¶

func (s *Store) DroppedCount() int64

DroppedCount returns the number of audit entries that have been dropped since process start. Used by ops endpoints and alerting.

func (*Store) EnqueueCtx ¶

func (s *Store) EnqueueCtx(ctx context.Context, entry *ToolCall)

EnqueueCtx adds a tool call to the write buffer with a request context for log correlation. Non-blocking; logs and counts any dropped entries via DroppedCount so operators can detect compliance gaps.

H3 fix (phase 2i): previously the worker-not-started fallback swallowed Record errors with `_ = s.Record(entry)` and the buffer-full path logged at Warn level with no counter. Both paths now go through incDropped so monitoring can alert on non-zero DroppedCount.

Sync-fallback path logs Error on every drop (rare — worker normally running). Buffer-full path logs a Warning every 100 drops with the cumulative count so a noisy backlog doesn't spam error logs but operators still see the trend.

ctx flows from the audit middleware (request-scoped) so trace IDs thread through to the persistence-failure log. nil ctx falls back to context.Background — preserves existing test-fixture callers.

func (*Store) GetGlobalStats ¶

func (s *Store) GetGlobalStats(since time.Time) (*Stats, error)

GetGlobalStats returns aggregate stats across all users since the given time.

func (*Store) GetOrderAttribution ¶

func (s *Store) GetOrderAttribution(email, orderID string) ([]*ToolCall, error)

GetOrderAttribution returns the tool call that placed the given order and all other tool calls from the same session in the 60 seconds before the order, providing a decision trace that shows how the AI arrived at the trade.

func (*Store) GetStats ¶

func (s *Store) GetStats(email string, since time.Time, category string, errorsOnly bool) (*Stats, error)

GetStats returns aggregate stats for a given email since the given time. If since is zero, all records are included. Optional category and errorsOnly filters scope the stats to match the user's active filters.

func (*Store) GetToolCounts ¶

func (s *Store) GetToolCounts(email string, since time.Time, category string, errorsOnly bool) (map[string]int, error)

GetToolCounts returns tool_name -> count for the given email since the given time. Results are ordered by count descending, limited to the top 20 tools. Optional category and errorsOnly filters scope results to match the user's active filters.

func (*Store) GetToolHistograms ¶

func (s *Store) GetToolHistograms(since time.Time) ([]ToolHistogram, error)

GetToolHistograms returns per-tool latency histograms for tool calls since the given time. Histograms are bucketed by MetricsBuckets; results exclude the synthetic __chain_break row (matches GetToolMetrics convention).

Implementation note: the SQL builds one CASE-binning column per bucket boundary. This is straightforward at 6 buckets; if the boundary list grows past ~15, the query should be split into per-bucket UNION ALL or moved to in-memory histograms.

func (*Store) GetToolMetrics ¶

func (s *Store) GetToolMetrics(since time.Time) ([]ToolMetric, error)

GetToolMetrics returns per-tool aggregate metrics (call count, avg/max latency, error count) for all tool calls since the given time. Results are ordered by call count descending, limited to the top 50 tools.

func (*Store) GetTopErrorUsers ¶

func (s *Store) GetTopErrorUsers(since time.Time, limit int) ([]UserErrorCount, error)

GetTopErrorUsers returns the top N users with the most errors since the given time. Email values are decrypted if an encryption key is configured.

func (*Store) InitTable ¶

func (s *Store) InitTable() error

InitTable creates the tool_calls table and indexes if they do not exist.

func (*Store) InvalidateStatsCache ¶

func (s *Store) InvalidateStatsCache(email string)

InvalidateStatsCache drops cached UserOrderStats entries for the given email. Called automatically on Record() for order-writing tools, but exposed publicly for tests and callers that bypass Record().

func (*Store) List ¶

func (s *Store) List(email string, opts ListOptions) ([]*ToolCall, int, error)

List retrieves tool call records for a given email, filtered and paginated according to opts. It returns the matching entries, the total count of matching records (ignoring limit/offset), and any error.

func (*Store) ListOrders ¶

func (s *Store) ListOrders(email string, since time.Time) ([]*ToolCall, error)

ListOrders returns tool calls with order IDs for the given email.

func (*Store) Record ¶

func (s *Store) Record(entry *ToolCall) error

Record inserts a tool call entry into the audit log. Duplicate call_ids are silently ignored (ON CONFLICT (call_id) DO NOTHING). When an encryption key is set, the email column stores the HMAC hash (for queryable lookups) and email_encrypted stores the AES-GCM ciphertext (for display/export). Legacy rows without encryption store plaintext.

SQL portability: ON CONFLICT (call_id) DO NOTHING is the dialect-portable idempotent-insert form (SQLite ≥3.24, Postgres ≥9.5) per kite-mcp-server Phase 2.1 audit. Single SQL string for both dialects, no runtime branching.

func (*Store) RemoveActivityListener ¶

func (s *Store) RemoveActivityListener(id string)

RemoveActivityListener unregisters a listener by id and closes its channel.

func (*Store) SeedChain ¶

func (s *Store) SeedChain()

SeedChain reads the last entry_hash from the database to resume the hash chain after a restart. Must be called after InitTable and SetEncryptionKey.

func (*Store) SetAnomalyCacheEventDispatcher ¶

func (s *Store) SetAnomalyCacheEventDispatcher(d *domain.EventDispatcher)

SetAnomalyCacheEventDispatcher wires the domain event dispatcher into the in-memory UserOrderStats cache so every baseline snapshot, user-scoped invalidation, and per-entry eviction is published as a typed domain.AnomalyCache*Event. Pure plumbing — the Store does not itself subscribe to these events. app/wire.go calls this once at startup; passing nil restores the legacy no-dispatch behaviour.

func (*Store) SetEncryptionKey ¶

func (s *Store) SetEncryptionKey(key []byte)

SetEncryptionKey configures the key used for HMAC email hashing, AES-GCM email encryption, and HMAC hash chaining. Call this before StartWorker.

func (*Store) SetLoggerPort ¶

func (s *Store) SetLoggerPort(l logport.Logger)

SetLoggerPort assigns a structured logger via the kc/logger.Logger port for background-worker diagnostics.

Nil is permitted: every internal logger call site is nil-checked (matches pre-migration semantics where unset s.logger was a no-op).

func (*Store) StartRetentionWorkerCtx ¶

func (s *Store) StartRetentionWorkerCtx(ctx context.Context, days int)

StartRetentionWorker launches a background goroutine that runs CleanupOldRecords(days) once every retentionTickInterval (default 24h).

When days <= 0 the worker is not started (retention disabled); callers can still invoke CleanupOldRecords directly if they want a one-shot cleanup.

This is intended as an in-package backstop that operators can enable when no external scheduler owns audit retention. The existing app/wire.go scheduler already runs DeleteOlderThan daily at 03:00 IST for production deployments; leaving the in-package worker inactive avoids double-deletion. Callers that want to drive retention entirely from the audit package (e.g. tests, alternative wiring, tooling) should call this method explicitly and pair it with StopRetentionWorker on shutdown.

Safe to call multiple times: a second call is a no-op while a worker is already running.

ctx is the parent service context — captured by the goroutine for log correlation (the goroutine's cleanup-success / cleanup-failure log entries flow through it). Shutdown is still driven by the stop channel (close via StopRetentionWorker); ctx is currently for logging only, but capturing it now positions us to honour ctx.Done() without a public-API change.

func (*Store) StartWorkerCtx ¶

func (s *Store) StartWorkerCtx(ctx context.Context)

StartWorkerCtx starts a background goroutine that drains the write channel. The ctx is the parent service context — typically the App startup ctx — and is captured on the Store as serviceCtx so the goroutine's logger calls can correlate with app-level traces.

Call Stop() to gracefully drain and close. ctx cancellation is NOT the shutdown trigger today (Stop() closes the channel directly), but a future enhancement could honour ctx.Done() to abort in-flight writes; capturing the ctx now positions us for that without a public-API change.

func (*Store) StatsCacheHitRate ¶

func (s *Store) StatsCacheHitRate() float64

StatsCacheHitRate exposes the UserOrderStats cache hit ratio for monitoring endpoints. Returns 0 before any queries have been made.

func (*Store) Stop ¶

func (s *Store) Stop()

Stop gracefully drains the buffer and waits for completion.

Stop is idempotent: calling it multiple times is safe and only the first call closes the worker channel. This guards against the case where more than one graceful-shutdown path (e.g. the HTTP signal handler and a test teardown) both try to Stop the same Store, which would otherwise produce `panic: close of closed channel`.

func (*Store) StopRetentionWorker ¶

func (s *Store) StopRetentionWorker()

StopRetentionWorker signals the retention goroutine to exit and blocks until it has finished. Safe to call when the worker was never started (no-op) and idempotent if called repeatedly.

func (*Store) UserOrderStats ¶

func (s *Store) UserOrderStats(email string, days int) (mean, stdev, count float64)

UserOrderStats returns the rolling mean and population standard deviation of a user's order values (qty * price) over the last `days` days. It also returns the raw count of rows found (regardless of the baseline floor) so callers can distinguish "unknown user" from "known user below threshold".

Semantics:

• Only `place_order` and `modify_order` rows are considered (the two tools where a user-triggered currency amount is actually at stake).
• Rows with missing/zero quantity or price are skipped (MARKET orders have no price at submission time and carry no useful value signal).
• When fewer than minBaselineOrders usable rows exist, mean and stdev are returned as zero. This is the "no baseline yet" sentinel.
• stdev uses the population formula (divide by N, not N-1). Population stdev is the correct choice here because we treat the observed window as the full sample of this user's trading behaviour for the purpose of anomaly detection — we are not inferring a population from a sample, we ARE the sample.

func (*Store) VerifyChain ¶

func (s *Store) VerifyChain() (*ChainVerification, error)

VerifyChain walks every entry in id order, recomputes HMAC hashes, and compares them with the stored entry_hash. Chain-break markers (tool_name = "__chain_break") are expected discontinuities and reset the expected prev_hash.