Backend Live Coding

-- ======================================================================
-- Q: You need text search over assets and alerts. PostgreSQL or
-- Elasticsearch/OpenSearch?
--
-- I start with PostgreSQL full-text search (tsvector + GIN) and pg_trgm for
-- typos/LIKE '%x%'. It covers many cases without a new system to maintain. I
-- move to OpenSearch/Elasticsearch when I need facets/aggregations over millions
-- of documents, relevance tuning, many languages or log-style search. Then
-- PostgreSQL stays the source of truth and the index is updated from events
-- (outbox → indexer), with the ability to do a full reindex.
-- ======================================================================

-- Generated column + GIN index
ALTER TABLE alerts ADD COLUMN search tsvector
  GENERATED ALWAYS AS (
    setweight(to_tsvector('simple', coalesce(title, '')), 'A') ||
    setweight(to_tsvector('simple', coalesce(description, '')), 'B')
  ) STORED;
CREATE INDEX CONCURRENTLY ix_alerts_search ON alerts USING GIN (search);

SELECT id, title, ts_rank(search, q) AS rank
FROM alerts, websearch_to_tsquery('simple', 'ssh brute force') q
WHERE tenant_id = 42 AND search @@ q
ORDER BY rank DESC LIMIT 20;

-- Typos / fuzzy
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE INDEX ix_assets_hostname_trgm ON assets USING GIN (hostname gin_trgm_ops);
SELECT hostname FROM assets WHERE hostname % 'webserv01' ORDER BY similarity(hostname, 'webserv01') DESC;

# ======================================================================
# Q: What scheduler do you use for periodic jobs and how do you avoid double
# execution?
#
# Celery beat (or APScheduler, or a Kubernetes CronJob). The scheduler runs on a
# single instance, but for safety (a deploy with two pods for a few seconds) I
# put a lock inside the job. The PostgreSQL advisory lock is a good choice if I
# don't want Redis:
# Each job records start/end/status in a job_runs table, so there is an alert if
# a job "did not run" (dead man's switch), not only if it failed.
# ======================================================================

from sqlalchemy import text

def run_exclusive(session, job_key: int, fn):
    got = session.scalar(text('SELECT pg_try_advisory_xact_lock(:k)'), {'k': job_key})
    if not got:
        return 'skipped: already running'
    return fn(session)        # the lock is released automatically at the end of the transaction

# ======================================================================
# Q: How do you reliably send an event when something changes in the database
# (e.g. "new critical alert")?
#
# With a transactional outbox. If I commit and then publish, and the publish
# fails, the event is lost. If I publish before the commit and the commit fails,
# I send an event for something that did not happen. That is why I write the
# event to an outbox table in the same transaction, and a relay worker sends it
# to the broker.
# Delivery is at-least-once, so consumers must be idempotent (dedup by event
# id). SKIP LOCKED is the classic pattern for a job queue inside PostgreSQL.
# ======================================================================

with session.begin():
    alert = Alert(tenant_id=tid, severity='critical', asset_id=aid)
    session.add(alert)
    session.flush()
    session.add(Outbox(topic='alerts.created',
                       key=str(alert.id),
                       payload={'alert_id': alert.id, 'tenant_id': tid}))

# relay worker (Celery beat every few seconds or a separate process)
def relay_outbox(session, batch=500):
    with session.begin():
        rows = session.scalars(
            select(Outbox).where(Outbox.sent_at.is_(None))
            .order_by(Outbox.id).limit(batch)
            .with_for_update(skip_locked=True)).all()   # many workers without conflicts
        for row in rows:
            broker.publish(row.topic, key=row.key, value=row.payload)
            row.sent_at = func.now()

# ======================================================================
# Q: How do you read 10 million rows from the database for an export without
# running out of memory?
#
# The export runs in a background job, writes to S3, and the user gets a link
# when it finishes. Never inside an HTTP request.
# ======================================================================

# SQLAlchemy: server-side cursor + yield_per
stmt = select(Event).where(Event.tenant_id == tid).order_by(Event.id)
for partition in session.execute(stmt.execution_options(yield_per=2000)).scalars().partitions():
    write_rows(partition)

# Django equivalent
for e in Event.objects.filter(tenant_id=tid).iterator(chunk_size=2000):
    write_row(e)

# ======================================================================
# Q: How do you design an enrichment pipeline (e.g. adding geo-IP, owner, CVE to
# each asset)?
#
# Stages with a clear responsibility: ingest → validate → normalize → enrich →
# store. Each stage is idempotent and records its state (status, attempts,
# last_error), so that only what failed is replayed. Enrichment from external
# APIs is done in batches, with a cache (I don't ask about the same IP twice), a
# rate limit and a circuit breaker.
# ======================================================================

import asyncio, httpx

async def enrich_ips(ips: list[str], concurrency: int = 20) -> dict[str, dict]:
    sem = asyncio.Semaphore(concurrency)         # limit on concurrent calls
    async with httpx.AsyncClient(timeout=5) as client:
        async def one(ip):
            if (hit := await cache.get(f'geo:{ip}')) is not None:
                return ip, hit
            async with sem:
                r = await client.get(f'{GEO_URL}/{ip}')
                r.raise_for_status()
                data = r.json()
            await cache.set(f'geo:{ip}', data, ex=86400)
            return ip, data
        results = await asyncio.gather(*(one(ip) for ip in set(ips)),
                                       return_exceptions=True)
    return {r[0]: r[1] for r in results if not isinstance(r, Exception)}