analytics

type Applier struct {
	// contains filtered or unexported fields
}

type ApplierOption func(*Applier)

type Backfiller struct {
	Snapshot SnapshotFunc
	Applier  *Applier
	Sink     Sink
	Batch    int // default 128
}

type Consumer struct {
	Group     string // "proj:analytics"
	Source    projection.Source
	Applier   *Applier
	Sink      Sink
	Upcasters *event.UpcasterChain // optional

	// OnApplied and OnFailure are optional observability hooks: OnApplied
	// fires once per envelope that is successfully applied to the Sink
	// (all writes succeeded); OnFailure fires once per envelope whose Sink
	// write returned an error. Neither fires on the skip/poison paths
	// (un-migratable payload, malformed payload, un-derivable tenant, or an
	// unmarked entity) — those are neither applied nor failed. Both are
	// nil-safe: callers that don't need metrics leave them unset. Kept as
	// bare func() (not a metrics dependency) so core/analytics stays free
	// of internal/metrics — callers wire prometheus.Counter.Inc directly.
	OnApplied func()
	OnFailure func()
}

type Event struct {
	TenantID  string          `json:"tenantId"`
	Aggregate string          `json:"aggregate"`
	AggID     string          `json:"aggId"`
	Version   int64           `json:"version"`
	Type      string          `json:"type"`
	Payload   json.RawMessage `json:"payload"` // already redacted by the applier
	At        time.Time       `json:"at"`
}

type Fact struct {
	TenantID  string          `json:"tenantId"`
	Aggregate string          `json:"aggregate"`
	AggID     string          `json:"aggId"`
	Version   int64           `json:"version"`
	Payload   json.RawMessage `json:"payload"` // already redacted by the applier
	At        time.Time       `json:"at"`
	Deleted   bool            `json:"deleted"`
}

type Reconciler struct {
	Snapshot SnapshotFunc
	Applier  *Applier
	Sink     Sink
}

type Report struct {
	Checked int `json:"checked"`
	Missing int `json:"missing"`
	Stale   int `json:"stale"`
	Healed  int `json:"healed"`
}

type Reprojector struct {
	Reg  *registry.Registry
	Sink Sink
	// Salt must match the deployment's Config.Analytics.HashSalt so re-projected
	// Hash fields hash identically to how the live applier wrote them.
	Salt string
}

type Sink interface {
	UpsertFacts(ctx context.Context, facts []Fact) error
	AppendEvents(ctx context.Context, events []Event) error
	Watermark(ctx context.Context, tenantID, aggregate, aggID string) (int64, error)
	SetWatermark(ctx context.Context, ws []Watermark) error
	// AllWatermarks returns every applied watermark for a tenant in one read —
	// the reconciler compares these against the source's current versions to
	// find drift (facts a skipped event left missing or stale) without a
	// per-aggregate round-trip.
	AllWatermarks(ctx context.Context, tenantID string) ([]Watermark, error)
	// LagByTenant reports per-tenant read-model freshness: for each tenant with
	// at least one fact, now() - (that tenant's newest fact commit time), in
	// seconds. An empty map means the sink holds no facts. Per-tenant (rather
	// than one fleet-wide number) so a single stalled tenant cannot hide behind
	// others still flowing; the poller derives the worst-case gauge and the
	// tenants-behind count from it without emitting a per-tenant metric series.
	LagByTenant(ctx context.Context) (map[string]float64, error)
	// ReprojectTenant re-writes stored fact AND event payloads for a tenant in
	// place, applying transform to each. aggregate "" means every aggregate.
	// It returns the number of rows whose payload actually changed. Used to
	// retroactively apply a tightened redaction allow-list to already-stored
	// data (a privacy correction): the transform re-projects each stored payload
	// through the entity's current AnalyticsSpec. Idempotent — a second run with
	// the same transform rewrites nothing.
	ReprojectTenant(ctx context.Context, tenantID, aggregate string, transform func(payload json.RawMessage) (json.RawMessage, error)) (rowsRewritten int64, err error)
	// PruneEvents deletes append-only history events committed strictly before
	// olderThan, across all tenants, and returns the count removed. It is the
	// retention control for the unbounded event log; facts (latest state) are
	// never pruned — only the history. Idempotent.
	PruneEvents(ctx context.Context, olderThan time.Time) (rowsDeleted int64, err error)
	// MaintainPartitions is the partition-management hook for sinks that
	// range-partition the event log: it creates upcoming partitions and, when
	// retention > 0, drops partitions entirely older than retention (instant
	// reclaim). Sinks that do not partition return (0, 0, nil).
	MaintainPartitions(ctx context.Context, retention time.Duration) (created, dropped int, err error)
	// PurgeTenant hard-deletes ALL of one tenant's co-located data — facts,
	// events, and watermarks — and returns the number of rows removed. It is the
	// erasure primitive for tenant offboarding and right-to-be-forgotten
	// requests: the ONE deliberately destructive operation on the sink, and the
	// only supported way to remove a tenant from the shared store. Idempotent
	// (purging an absent tenant deletes nothing and returns 0). Never invoked
	// automatically — an explicit operator step, like dropping a tenant database.
	PurgeTenant(ctx context.Context, tenantID string) (rowsDeleted int64, err error)
	Close() error
}

type SnapshotFunc func(ctx context.Context, tenantID string, fn func(event.Envelope) error) error

type Watermark struct {
	TenantID  string `json:"tenantId"`
	Aggregate string `json:"aggregate"`
	AggID     string `json:"aggId"`
	Version   int64  `json:"version"`
}