Storage

The Storage layer writes enriched events to PostgreSQL using batched executemany INSERT statements with intra-batch deduplication merge. The separate copy_events() method provides a COPY bulk-loading path used for very large historical files.

---

Write Path

flowchart LR
  BATCH[Enriched Batch] --> MERGE[Intra-Batch Merge<br/>dedup_key]
  MERGE --> INS[executemany INSERT<br/>events + templates]
  INS -->|Success| PG[(PostgreSQL<br/>events table)]
  INS -->|Fail| BUFFER[In-memory buffer<br/>MAX_BUFFER_SIZE 10 000]
  BUFFER -->|Overflow| DLQ[DLQ Topic + dlq_events]
  PG --> PUBSUB[Redis PUBLISH<br/>events:new]

---

Intra-Batch Merge

Before writing, merge events with same dedup_key within the batch:

# storage.py
def _merge_batch(rows: list[dict]) -> list[dict]:
    merged = {}
    for row in rows:
        key = row.get("dedup_key")
        if key and key in merged:
            merged[key]["occurrence_count"] += row.get("occurrence_count", 1)
        else:
            merged[key] = row
    return list(merged.values())

---

PostgreSQL Batch INSERT

The hot write path is an executemany INSERT (events) followed by template upserts. occurrence_count is fed by the intra-batch merge via occurrence_delta:

async def write_batch(self, rows: list[dict]):
    if not rows:
        return

    collapsed = _collapse_duplicates_in_batch(rows)

    with self._conn.cursor() as cur:
        cur.executemany(_INSERT_EVENT_SQL, collapsed)
        template_rows = [r for r in collapsed
                         if r.get("template_id") and r.get("template_text")]
        if template_rows:
            cur.executemany(_INSERT_TEMPLATE_SQL, template_rows)
    self._conn.commit()
    _notify_redis(collapsed)

INSERT INTO events (...) has no ON CONFLICT — the dedup backstop is the partial unique index ux_events_dedup_key (see Indexes). Templates are upserted with ON CONFLICT (template_id) DO UPDATE.

Bulk COPY Path

copy_events() is a separate method for bulk historical ingestion (e.g. large files), using COPY ... FROM STDIN. It is not the per-batch write path.

async def copy_events(self, events: list[dict]):
    with self._conn.cursor() as cur:
        with cur.copy("COPY events (...) FROM STDIN") as copy:
            for e in _collapse_duplicates_in_batch(events):
                copy.write_row((...))
    self._conn.commit()

---

Retry Logic (in-memory buffer)

When a write fails, the batch is buffered in memory and retried on the next successful write. If Postgres stays down and the buffer exceeds MAX_BUFFER_SIZE (10,000 rows), the oldest rows are offloaded to the DLQ to avoid OOM:

Attempt	Action
1	Write batch via executemany
Fail	Append to in-memory buffer
Buffer full	Oldest rows → DLQ (buffer_overflow_dropped)
Next write OK	Flush buffer first, then write new batch

---

Real-time Notification

After successful write:

# Notify dashboard via Redis pub/sub
await self.redis.publish("events:new", orjson.dumps({
    "count": len(rows),
    "timestamp": datetime.utcnow().isoformat()
}))

Frontend WebSocket receives this for live updates.

---

DLQ Persistence

Failed rows written to both: 1. Kafka dlq topic — for replay 2. PostgreSQL dlq_events table — for querying

CREATE TABLE dlq_events (
    id BIGSERIAL PRIMARY KEY,
    reason TEXT NOT NULL,
    raw_payload JSONB NOT NULL,
    received_at TIMESTAMPTZ DEFAULT NOW()
);

---

Events Table Schema

CREATE TABLE events (
    id BIGSERIAL,
    schema_version TEXT NOT NULL DEFAULT '1.0',
    timestamp TIMESTAMPTZ NOT NULL DEFAULT now(),
    source VARCHAR(255) NOT NULL DEFAULT 'legacy',
    environment VARCHAR(50) NOT NULL DEFAULT 'production',
    type VARCHAR(100) NOT NULL DEFAULT 'application',
    severity VARCHAR(20) NOT NULL DEFAULT 'info',
    message TEXT NOT NULL DEFAULT '',
    summary TEXT,
    template_id TEXT,
    host VARCHAR(255) NOT NULL DEFAULT '',
    pod_name VARCHAR(255),
    trace_id VARCHAR(255),
    span_id VARCHAR(255),
    occurrence_count INT NOT NULL DEFAULT 1,
    tags JSONB NOT NULL DEFAULT '{}',
    dedup_key VARCHAR(255),
    severity_rank SMALLINT NOT NULL DEFAULT 0,
    is_error_or_worse BOOLEAN NOT NULL DEFAULT FALSE,
    priority VARCHAR(2),
    priority_confidence FLOAT,
    priority_source VARCHAR(20),
    model_version VARCHAR(100),
    inserted_at TIMESTAMPTZ NOT NULL DEFAULT now()
) PARTITION BY RANGE (timestamp);

There is no REFERENCES constraint on template_id (it is a plain TEXT column), and the column is not nullable.

---

Indexes

Index	Columns	Purpose
ix_events_timestamp	timestamp DESC	Recent queries
ix_events_source_sev_ts	source, severity, timestamp DESC	Filter by source+severity
ix_events_severity	severity	Severity breakdown
ix_events_priority	priority WHERE priority IS NOT NULL	Priority dashboards
ix_events_priority_ts	priority, timestamp DESC WHERE priority IS NOT NULL	Priority + time range
ix_events_template_id	template_id	Template analytics
ix_events_message_trgm	message (GIN pg_trgm)	Full-text search
ux_events_id	(id, timestamp)	Primary key (partition-aware)
ux_events_dedup_key	(dedup_key, timestamp) WHERE dedup_key IS NOT NULL	Deduplication backstop

---

Monitoring

# Write rate
docker compose logs consumer-pipeline | grep written_total

# Batch timing
docker compose logs consumer-pipeline | grep processing_time_ms

# DLQ count
curl -s http://localhost:8000/api/kpis | jq .dlq_count

---

Troubleshooting

Issue	Check
Batch insert fails	Check docker compose logs consumer-pipeline for constraint violations
High DLQ	Inspect dlq_events table for reason patterns
Slow writes	Check PG pg_stat_activity, increase work_mem
Duplicate events	Verify dedup_key generation, check Redis dedup