AHU AI Observatory — Backend (P1) Implementation Plan

cmd/observatory/main.go        wiring: config → pool → migrator → ingester + API server
internal/config/config.go      YAML config: pg dsn, redis, stream/group, listen, role tokens, retention
internal/store/migrate.go      embedded SQL migrations, applied on boot (idempotent)
internal/store/migrations/*.sql
internal/store/calls.go        metadata inserts (dedup) + query helpers (summary/series/calls)
internal/store/bodies.go       hash-chained body inserts (advisory-lock serialized), reads, chain verify
internal/store/access.go       body_access_audit writes + reads
internal/ingest/ingester.go    consumer-group loop, parse, tx insert, ack, autoclaim, metrics
internal/api/server.go         mux, auth middleware (token→role), handlers, healthz, metrics
deploy/observatory.example.yaml
deploy/compose.yaml            observatory + timescaledb; joins external ahu-platform_default network
Dockerfile
scripts/testdb.sh              throwaway timescaledb for tests (port 5434)

type Role string // "executive" | "operator" | "auditor"
type Config struct {
    Listen        string            `yaml:"listen"`         // default ":8300"
    PGDSN         string            `yaml:"pg_dsn"`
    RedisURL      string            `yaml:"redis_url"`
    StreamKey     string            `yaml:"stream_key"`     // default "ahu.ai.audit"
    ConsumerGroup string            `yaml:"consumer_group"` // default "observatory"
    ConsumerName  string            `yaml:"consumer_name"`  // default hostname; fallback "obs-1"
    Tokens        map[string]Role   `yaml:"tokens"`         // bearer token -> role
    RetentionDays int               `yaml:"retention_days"` // metadata retention, default 400 (~13 months)
}

cd /home/efran/remote-development/poc-ahu-ai/ahu-ai-observatory
git checkout -b feat/observatory-p1 2>/dev/null || git checkout feat/observatory-p1
go mod init datahive.id/ahu-ai-observatory

package config

import (
    "os"
    "path/filepath"
    "testing"
)

func write(t *testing.T, s string) string {
    t.Helper()
    p := filepath.Join(t.TempDir(), "o.yaml")
    if err := os.WriteFile(p, []byte(s), 0o644); err != nil {
        t.Fatal(err)
    }
    return p
}

func TestLoadValidAndDefaults(t *testing.T) {
    c, err := Load(write(t, `
pg_dsn: "postgres://u:p@h:5432/db"
redis_url: "redis://r:6379/0"
tokens:
  tok-exec: executive
  tok-aud: auditor
`))
    if err != nil {
        t.Fatal(err)
    }
    if c.Listen != ":8300" || c.StreamKey != "ahu.ai.audit" || c.ConsumerGroup != "observatory" {
        t.Fatalf("defaults wrong: %+v", c)
    }
    if c.RetentionDays != 400 || c.ConsumerName == "" {
        t.Fatalf("defaults wrong: %+v", c)
    }
    if c.Tokens["tok-aud"] != "auditor" {
        t.Fatalf("tokens: %+v", c.Tokens)
    }
}

func TestLoadRejectsBadRoleAndMissingDSN(t *testing.T) {
    if _, err := Load(write(t, "pg_dsn: x\nredis_url: y\ntokens: {t: superuser}")); err == nil {
        t.Fatal("bad role accepted")
    }
    if _, err := Load(write(t, "redis_url: y")); err == nil {
        t.Fatal("missing pg_dsn accepted")
    }
}

package config

import (
    "fmt"
    "os"

    "gopkg.in/yaml.v3"
)

type Role string

const (
    RoleExecutive Role = "executive"
    RoleOperator  Role = "operator"
    RoleAuditor   Role = "auditor"
)

type Config struct {
    Listen        string          `yaml:"listen"`
    PGDSN         string          `yaml:"pg_dsn"`
    RedisURL      string          `yaml:"redis_url"`
    StreamKey     string          `yaml:"stream_key"`
    ConsumerGroup string          `yaml:"consumer_group"`
    ConsumerName  string          `yaml:"consumer_name"`
    Tokens        map[string]Role `yaml:"tokens"`
    RetentionDays int             `yaml:"retention_days"`
}

func Load(path string) (*Config, error) {
    b, err := os.ReadFile(path)
    if err != nil {
        return nil, err
    }
    var c Config
    if err := yaml.Unmarshal(b, &c); err != nil {
        return nil, err
    }
    if c.PGDSN == "" || c.RedisURL == "" {
        return nil, fmt.Errorf("pg_dsn and redis_url are required")
    }
    if c.Listen == "" {
        c.Listen = ":8300"
    }
    if c.StreamKey == "" {
        c.StreamKey = "ahu.ai.audit"
    }
    if c.ConsumerGroup == "" {
        c.ConsumerGroup = "observatory"
    }
    if c.ConsumerName == "" {
        if h, err := os.Hostname(); err == nil && h != "" {
            c.ConsumerName = h
        } else {
            c.ConsumerName = "obs-1"
        }
    }
    if c.RetentionDays == 0 {
        c.RetentionDays = 400
    }
    for tok, r := range c.Tokens {
        switch r {
        case RoleExecutive, RoleOperator, RoleAuditor:
        default:
            return nil, fmt.Errorf("token %q: invalid role %q", tok, r)
        }
    }
    return &c, nil
}

go get gopkg.in/yaml.v3 && go test ./internal/config/ -v
git add go.mod go.sum internal/config/ .gitignore docs/
git commit -m "feat(config): observatory config loader with role validation

Claude-Session: https://claude.ai/code/session_01Cco5bXPQVNVk342NEW1pdT"

#!/usr/bin/env bash
# Throwaway TimescaleDB for observatory tests (port 5434 — 5433 is taken by the OCR stack Postgres). Usage: ./scripts/testdb.sh [stop]
set -euo pipefail
if [ "${1:-}" = "stop" ]; then docker rm -f obs-test-pg >/dev/null 2>&1 || true; exit 0; fi
docker rm -f obs-test-pg >/dev/null 2>&1 || true
docker run -d --name obs-test-pg -p 5434:5432 -e POSTGRES_PASSWORD=test timescale/timescaledb:latest-pg16 >/dev/null
for i in $(seq 1 30); do
  docker exec obs-test-pg pg_isready -U postgres >/dev/null 2>&1 && { echo "ready: postgres://postgres:test@localhost:5434/postgres"; exit 0; }
  sleep 1
done
echo "testdb failed to start" >&2; exit 1

package store

import (
    "context"
    "testing"
)

func TestMigrateIdempotentAndSchemaShape(t *testing.T) {
    pool := TestPool(t)
    ctx := context.Background()
    if err := Migrate(ctx, pool); err != nil { // second run (TestPool already ran it once)
        t.Fatalf("re-migrate not idempotent: %v", err)
    }
    var isHyper bool
    err := pool.QueryRow(ctx,
        `SELECT count(*) = 1 FROM timescaledb_information.hypertables WHERE hypertable_name = 'ai_calls'`).Scan(&isHyper)
    if err != nil || !isHyper {
        t.Fatalf("ai_calls not a hypertable: %v %v", isHyper, err)
    }
    var n int
    if err := pool.QueryRow(ctx,
        `SELECT count(*) FROM timescaledb_information.continuous_aggregates WHERE view_name = 'ai_calls_1m'`).Scan(&n); err != nil || n != 1 {
        t.Fatalf("continuous aggregate missing: %d %v", n, err)
    }
    for _, tbl := range []string{"ai_call_bodies", "body_access_audit"} {
        if err := pool.QueryRow(ctx, `SELECT count(*) FROM `+tbl).Scan(&n); err != nil {
            t.Fatalf("table %s missing: %v", tbl, err)
        }
    }
}

CREATE EXTENSION IF NOT EXISTS timescaledb;

CREATE TABLE IF NOT EXISTS ai_calls (
  event_id       text        NOT NULL,
  ts             timestamptz NOT NULL,
  engine         text NOT NULL DEFAULT '',
  surface        text NOT NULL DEFAULT '',
  user_id        text NOT NULL DEFAULT '',
  trace_id       text NOT NULL DEFAULT '',
  upstream       text NOT NULL DEFAULT '',
  model          text NOT NULL DEFAULT '',
  upstream_class text NOT NULL DEFAULT '',
  traffic_class  text NOT NULL DEFAULT '',
  operation      text NOT NULL DEFAULT '',
  status         text NOT NULL DEFAULT '',
  error_code     text NOT NULL DEFAULT '',
  queue_ms       bigint NOT NULL DEFAULT 0,
  upstream_ms    bigint NOT NULL DEFAULT 0,
  total_ms       bigint NOT NULL DEFAULT 0,
  tokens_in      int,
  tokens_out     int,
  doc_hash       text NOT NULL DEFAULT '',
  pages          int,
  job_id         text NOT NULL DEFAULT '',
  parent_job_id  text NOT NULL DEFAULT '',
  config_version text NOT NULL DEFAULT '',
  body_ref       bigint,
  body_truncated boolean NOT NULL DEFAULT false,
  ingested_at    timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (event_id, ts)
);
SELECT create_hypertable('ai_calls', 'ts', chunk_time_interval => INTERVAL '1 day', if_not_exists => TRUE);
CREATE INDEX IF NOT EXISTS ai_calls_engine_ts ON ai_calls (engine, ts DESC);
CREATE INDEX IF NOT EXISTS ai_calls_trace ON ai_calls (trace_id);

CREATE MATERIALIZED VIEW IF NOT EXISTS ai_calls_1m
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 minute', ts) AS bucket,
       engine, upstream, traffic_class, status,
       count(*)                AS calls,
       sum(coalesce(tokens_in,0))  AS tokens_in,
       sum(coalesce(tokens_out,0)) AS tokens_out,
       avg(total_ms)           AS avg_total_ms,
       max(total_ms)           AS max_total_ms,
       avg(queue_ms)           AS avg_queue_ms
FROM ai_calls GROUP BY bucket, engine, upstream, traffic_class, status
WITH NO DATA;
SELECT add_continuous_aggregate_policy('ai_calls_1m',
  start_offset => INTERVAL '2 hours', end_offset => INTERVAL '1 minute',
  schedule_interval => INTERVAL '1 minute', if_not_exists => TRUE);

CREATE TABLE IF NOT EXISTS ai_call_bodies (
  chain_pos         bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  event_id          text NOT NULL UNIQUE,
  ts                timestamptz NOT NULL,
  request_body_zst  bytea,
  response_body_zst bytea,
  full_body_sha256  text NOT NULL DEFAULT '',
  prev_hash         bytea NOT NULL,
  row_hash          bytea NOT NULL
);

CREATE OR REPLACE FUNCTION ai_call_bodies_immutable() RETURNS trigger AS $$
BEGIN RAISE EXCEPTION 'ai_call_bodies is append-only'; END;
$$ LANGUAGE plpgsql;
DROP TRIGGER IF EXISTS bodies_no_update ON ai_call_bodies;
CREATE TRIGGER bodies_no_update BEFORE UPDATE OR DELETE ON ai_call_bodies
  FOR EACH ROW EXECUTE FUNCTION ai_call_bodies_immutable();
DROP TRIGGER IF EXISTS bodies_no_truncate ON ai_call_bodies;
CREATE TRIGGER bodies_no_truncate BEFORE TRUNCATE ON ai_call_bodies
  FOR EACH STATEMENT EXECUTE FUNCTION ai_call_bodies_immutable();

CREATE TABLE IF NOT EXISTS body_access_audit (
  id       bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  at       timestamptz NOT NULL DEFAULT now(),
  actor    text NOT NULL,
  role     text NOT NULL,
  event_id text NOT NULL,
  action   text NOT NULL
);

package store

import (
    "context"
    "embed"
    "fmt"
    "sort"

    "github.com/jackc/pgx/v5/pgxpool"
)

//go:embed migrations/*.sql
var migrationsFS embed.FS

func Migrate(ctx context.Context, pool *pgxpool.Pool) error {
    entries, err := migrationsFS.ReadDir("migrations")
    if err != nil {
        return err
    }
    names := make([]string, 0, len(entries))
    for _, e := range entries {
        names = append(names, e.Name())
    }
    sort.Strings(names)
    for _, n := range names {
        sql, err := migrationsFS.ReadFile("migrations/" + n)
        if err != nil {
            return err
        }
        if _, err := pool.Exec(ctx, string(sql)); err != nil {
            return fmt.Errorf("migration %s: %w", n, err)
        }
    }
    return nil
}

package store

import (
    "context"
    "os"
    "testing"

    "github.com/jackc/pgx/v5/pgxpool"
)

func TestPool(t *testing.T) *pgxpool.Pool {
    t.Helper()
    dsn := os.Getenv("TEST_PG_DSN")
    if dsn == "" {
        dsn = "postgres://postgres:test@localhost:5434/postgres"
    }
    pool, err := pgxpool.New(context.Background(), dsn)
    if err == nil {
        err = pool.Ping(context.Background())
    }
    if err != nil {
        t.Skipf("test DB unavailable (run ./scripts/testdb.sh): %v", err)
    }
    if err := Migrate(context.Background(), pool); err != nil {
        t.Fatalf("migrate: %v", err)
    }
    t.Cleanup(pool.Close)
    return pool
}

go get github.com/jackc/pgx/v5 && go test ./internal/store/ -v
git add internal/store/ scripts/
git commit -m "feat(store): timescale schema, idempotent migrator, test harness

Claude-Session: https://claude.ai/code/session_01Cco5bXPQVNVk342NEW1pdT"

type Body struct {
    EventID        string
    TS             time.Time
    RequestZst     []byte
    ResponseZst    []byte
    FullBodySHA256 string
}
// InsertBody appends within tx, serialized by pg advisory xact lock 4711.
// Returns chain_pos. Duplicate event_id returns (0, ErrDuplicate).
func InsertBody(ctx context.Context, tx pgx.Tx, b Body) (int64, error)
func GetBody(ctx context.Context, pool *pgxpool.Pool, eventID string) (*StoredBody, error) // ErrNotFound
type StoredBody struct{ ChainPos int64; Body; PrevHash, RowHash []byte }
// VerifyChain recomputes hashes over [fromPos,toPos]; returns first bad pos or 0.
func VerifyChain(ctx context.Context, pool *pgxpool.Pool, fromPos, toPos int64) (int64, error)
var ErrDuplicate = errors.New("duplicate body")
var ErrNotFound = errors.New("not found")

package store

import (
    "context"
    "crypto/sha256"
    "testing"
    "time"

    "github.com/oklog/ulid/v2"
)

func insertOne(t *testing.T, pool *pgxpool.Pool, req, resp []byte) (string, int64) {
    t.Helper()
    ctx := context.Background()
    id := ulid.Make().String()
    tx, err := pool.Begin(ctx)
    if err != nil {
        t.Fatal(err)
    }
    pos, err := InsertBody(ctx, tx, Body{EventID: id, TS: time.Now(), RequestZst: req, ResponseZst: resp})
    if err != nil {
        t.Fatal(err)
    }
    if err := tx.Commit(ctx); err != nil {
        t.Fatal(err)
    }
    return id, pos
}

func TestChainLinksAndVerifies(t *testing.T) {
    pool := TestPool(t)
    ctx := context.Background()
    _, p1 := insertOne(t, pool, []byte("req1"), []byte("resp1"))
    id2, p2 := insertOne(t, pool, []byte("req2"), nil)
    if p2 <= p1 {
        t.Fatalf("chain positions not increasing: %d %d", p1, p2)
    }
    b2, err := GetBody(ctx, pool, id2)
    if err != nil {
        t.Fatal(err)
    }
    // b2.prev_hash must equal row_hash of the row before it in the chain
    var prevRow []byte
    if err := pool.QueryRow(ctx,
        `SELECT row_hash FROM ai_call_bodies WHERE chain_pos < $1 ORDER BY chain_pos DESC LIMIT 1`,
        b2.ChainPos).Scan(&prevRow); err != nil {
        t.Fatal(err)
    }
    if string(b2.PrevHash) != string(prevRow) {
        t.Fatal("prev_hash does not link to predecessor row_hash")
    }
    reqH := sha256.Sum256([]byte("req2"))
    respH := sha256.Sum256(nil)
    want := sha256.Sum256(append(append(append(append([]byte{}, b2.PrevHash...), []byte(id2)...), reqH[:]...), respH[:]...))
    if string(b2.RowHash) != string(want[:]) {
        t.Fatal("row_hash formula mismatch")
    }
    if bad, err := VerifyChain(ctx, pool, 0, 0); err != nil || bad != 0 {
        t.Fatalf("verify: bad=%d err=%v", bad, err)
    }
}

func TestDuplicateAndImmutability(t *testing.T) {
    pool := TestPool(t)
    ctx := context.Background()
    id, pos := insertOne(t, pool, []byte("x"), []byte("y"))
    tx, _ := pool.Begin(ctx)
    if _, err := InsertBody(ctx, tx, Body{EventID: id, TS: time.Now()}); err != ErrDuplicate {
        t.Fatalf("want ErrDuplicate, got %v", err)
    }
    tx.Rollback(ctx)
    if _, err := pool.Exec(ctx, `UPDATE ai_call_bodies SET full_body_sha256='hax' WHERE chain_pos=$1`, pos); err == nil {
        t.Fatal("UPDATE not blocked")
    }
    if _, err := pool.Exec(ctx, `DELETE FROM ai_call_bodies WHERE chain_pos=$1`, pos); err == nil {
        t.Fatal("DELETE not blocked")
    }
}

package store

import (
    "context"
    "crypto/sha256"
    "errors"
    "time"

    "github.com/jackc/pgx/v5"
    "github.com/jackc/pgx/v5/pgxpool"
)

var (
    ErrDuplicate = errors.New("duplicate body")
    ErrNotFound  = errors.New("not found")
)

type Body struct {
    EventID        string
    TS             time.Time
    RequestZst     []byte
    ResponseZst    []byte
    FullBodySHA256 string
}

type StoredBody struct {
    ChainPos int64
    Body
    PrevHash []byte
    RowHash  []byte
}

func rowHash(prev []byte, eventID string, req, resp []byte) []byte {
    rq := sha256.Sum256(req)
    rs := sha256.Sum256(resp)
    h := sha256.New()
    h.Write(prev)
    h.Write([]byte(eventID))
    h.Write(rq[:])
    h.Write(rs[:])
    return h.Sum(nil)
}

func InsertBody(ctx context.Context, tx pgx.Tx, b Body) (int64, error) {
    // Serialize chain appends: xact-scoped advisory lock releases on commit/rollback.
    if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(4711)`); err != nil {
        return 0, err
    }
    var dup bool
    if err := tx.QueryRow(ctx,
        `SELECT EXISTS(SELECT 1 FROM ai_call_bodies WHERE event_id=$1)`, b.EventID).Scan(&dup); err != nil {
        return 0, err
    }
    if dup {
        return 0, ErrDuplicate
    }
    prev := make([]byte, 32)
    err := tx.QueryRow(ctx,
        `SELECT row_hash FROM ai_call_bodies ORDER BY chain_pos DESC LIMIT 1`).Scan(&prev)
    if err != nil && err != pgx.ErrNoRows {
        return 0, err
    }
    rh := rowHash(prev, b.EventID, b.RequestZst, b.ResponseZst)
    var pos int64
    err = tx.QueryRow(ctx, `
        INSERT INTO ai_call_bodies (event_id, ts, request_body_zst, response_body_zst, full_body_sha256, prev_hash, row_hash)
        VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING chain_pos`,
        b.EventID, b.TS, b.RequestZst, b.ResponseZst, b.FullBodySHA256, prev, rh).Scan(&pos)
    return pos, err
}

func GetBody(ctx context.Context, pool *pgxpool.Pool, eventID string) (*StoredBody, error) {
    var s StoredBody
    err := pool.QueryRow(ctx, `
        SELECT chain_pos, event_id, ts, request_body_zst, response_body_zst, full_body_sha256, prev_hash, row_hash
        FROM ai_call_bodies WHERE event_id=$1`, eventID).
        Scan(&s.ChainPos, &s.EventID, &s.TS, &s.RequestZst, &s.ResponseZst, &s.FullBodySHA256, &s.PrevHash, &s.RowHash)
    if err == pgx.ErrNoRows {
        return nil, ErrNotFound
    }
    return &s, err
}

// VerifyChain returns the first chain_pos whose row_hash does not verify, or 0 if all good.
// fromPos/toPos of 0 mean unbounded.
func VerifyChain(ctx context.Context, pool *pgxpool.Pool, fromPos, toPos int64) (int64, error) {
    q := `SELECT chain_pos, event_id, request_body_zst, response_body_zst, prev_hash, row_hash
          FROM ai_call_bodies WHERE ($1 = 0 OR chain_pos >= $1) AND ($2 = 0 OR chain_pos <= $2)
          ORDER BY chain_pos`
    rows, err := pool.Query(ctx, q, fromPos, toPos)
    if err != nil {
        return 0, err
    }
    defer rows.Close()
    var prev []byte
    first := true
    for rows.Next() {
        var pos int64
        var id string
        var req, resp, ph, rh []byte
        if err := rows.Scan(&pos, &id, &req, &resp, &ph, &rh); err != nil {
            return 0, err
        }
        if first && fromPos > 1 {
            prev = ph // mid-chain start: trust the stored prev_hash as anchor
        }
        if first && fromPos <= 1 {
            prev = make([]byte, 32)
        }
        first = false
        if string(ph) != string(prev) {
            return pos, nil
        }
        if string(rowHash(prev, id, req, resp)) != string(rh) {
            return pos, nil
        }
        prev = rh
    }
    return 0, rows.Err()
}

go get github.com/oklog/ulid/v2 && go test ./internal/store/ -v -race
git add internal/store/ go.mod go.sum
git commit -m "feat(store): hash-chained append-only body store with verification

Claude-Session: https://claude.ai/code/session_01Cco5bXPQVNVk342NEW1pdT"

// Event mirrors the gateway's audit schema v1 JSON exactly (see
// ../ahu-gpu-manager/internal/audit/event.go). []byte fields carry zstd bytes
// (encoding/json base64-decodes them automatically on unmarshal).
type Event struct {
    EventID string `json:"event_id"`
    SchemaVer int  `json:"schema_ver"`
    TS      string `json:"ts"`
    Engine  string `json:"engine"`
    Surface string `json:"surface"`
    UserID  string `json:"user_id"`
    TraceID string `json:"trace_id"`
    Upstream string `json:"upstream"`
    Model   string `json:"model"`
    UpstreamClass string `json:"upstream_class"`
    TrafficClass  string `json:"traffic_class"`
    Operation string `json:"operation"`
    Status    string `json:"status"`
    ErrorCode string `json:"error_code"`
    QueueMs   int64  `json:"queue_ms"`
    UpstreamMs int64 `json:"upstream_ms"`
    TotalMs   int64  `json:"total_ms"`
    TokensIn  *int   `json:"tokens_in"`
    TokensOut *int   `json:"tokens_out"`
    DocHash   string `json:"doc_hash"`
    Pages     *int   `json:"pages"`
    JobID     string `json:"job_id"`
    ParentJobID string `json:"parent_job_id"`
    ConfigVersion string `json:"config_version"`
    RequestBodyZst  []byte `json:"request_body_zst"`
    ResponseBodyZst []byte `json:"response_body_zst"`
    BodyTruncated   bool   `json:"body_truncated"`
    FullBodySHA256  string `json:"full_body_sha256"`
}
// InsertEvent: one tx — body row (when bodies present) then metadata with
// body_ref; dedup on event_id (bodies dup OR metadata conflict → inserted=false, nil error).
func InsertEvent(ctx context.Context, pool *pgxpool.Pool, ev Event) (inserted bool, err error)
type CallFilter struct{ From, To time.Time; Engine, Upstream, Status, TraceID string; Limit int; BeforeTS time.Time; BeforeID string }
func QueryCalls(ctx, pool, f CallFilter) ([]CallRow, error)     // metadata list, ts DESC, keyset pagination
func QuerySummary(ctx, pool, from, to time.Time) ([]SummaryRow, error)   // totals by engine+status (raw table)
func QuerySeries(ctx, pool, from, to time.Time, bucket time.Duration, groupBy string) ([]SeriesRow, error) // groupBy ∈ engine|upstream|traffic_class (validated)
func LogBodyAccess(ctx, pool, actor, role, eventID, action string) error
func QueryBodyAccess(ctx, pool, limit int) ([]AccessRow, error)

func InsertEvent(ctx context.Context, pool *pgxpool.Pool, ev Event) (bool, error) {
    ts, err := time.Parse(time.RFC3339Nano, ev.TS)
    if err != nil {
        ts = time.Now().UTC()
        badTSCount.Add(1)
    }
    tx, err := pool.Begin(ctx)
    if err != nil {
        return false, err
    }
    defer tx.Rollback(ctx)
    var bodyRef *int64
    if len(ev.RequestBodyZst) > 0 || len(ev.ResponseBodyZst) > 0 {
        pos, err := InsertBody(ctx, tx, Body{EventID: ev.EventID, TS: ts,
            RequestZst: ev.RequestBodyZst, ResponseZst: ev.ResponseBodyZst, FullBodySHA256: ev.FullBodySHA256})
        switch err {
        case nil:
            bodyRef = &pos
        case ErrDuplicate:
            return false, tx.Commit(ctx) // whole event is a redelivery
        default:
            return false, err
        }
    }
    ct, err := tx.Exec(ctx, `INSERT INTO ai_calls (...27 cols...) VALUES (...) ON CONFLICT (event_id, ts) DO NOTHING`, /* all fields, bodyRef */)
    if err != nil {
        return false, err
    }
    if err := tx.Commit(ctx); err != nil {
        return false, err
    }
    return ct.RowsAffected() == 1, nil
}

func New(rdb redis.UniversalClient, pool *pgxpool.Pool, streamKey, group, consumer string, reg prometheus.Registerer) *Ingester
func (i *Ingester) Run(ctx context.Context) error // returns when ctx done
// Metrics on reg: obs_ingested_total, obs_deduped_total, obs_parse_errors_total,
// obs_pending gauge (XPENDING count, refreshed each loop)

listen: ":8300"
pg_dsn: "postgres://observatory:observatory@ahu-observatory-db:5432/observatory"
redis_url: "redis://ahu-platform-redis:6379/0"
stream_key: "ahu.ai.audit"
consumer_group: "observatory"
retention_days: 400
tokens:
  # replace before real use; roles: executive | operator | auditor
  tok-exec-CHANGEME: executive
  tok-ops-CHANGEME: operator
  tok-aud-CHANGEME: auditor

name: ahu-observatory
services:
  observatory:
    build: ..
    restart: unless-stopped
    ports: ["8300:8300"]
    environment:
      OBSERVATORY_CONFIG: /etc/ahu-observatory/observatory.yaml
    volumes:
      - ./observatory.yaml:/etc/ahu-observatory/observatory.yaml:ro
    depends_on: [db]
    networks: [default, platform]
  db:
    image: timescale/timescaledb:latest-pg16
    container_name: ahu-observatory-db
    restart: unless-stopped
    environment:
      POSTGRES_USER: observatory
      POSTGRES_PASSWORD: observatory
      POSTGRES_DB: observatory
    volumes: [obs-pg-data:/var/lib/postgresql/data]
    networks: [default]
networks:
  platform:
    external: true
    name: ahu-platform_default
volumes:
  obs-pg-data: