Data Organization and Infrastructure Challenges Questions

import asyncio, aiohttp, time, json, os, collections
from hashlib import blake2b

# config
MAX_QUEUE = 1000
BATCH_SIZE = 50
FLUSH_INTERVAL = 1.0
AGG_URL = "https://metrics.example/v1/ingest"
SPILL_DIR = "/tmp/metric_spill"

queue = asyncio.Queue(maxsize=MAX_QUEUE)
drop_counter = 0

def fingerprint_input(inp):
    h = blake2b(digest_size=8)
    h.update(json.dumps(inp, sort_keys=True).encode())
    return h.hexdigest()

# called in inference path (synchronous/non-blocking)
def record(inference_id, start_ts, end_ts, success, input_payload):
    evt = {
        "id": inference_id,
        "latency_ms": int((end_ts - start_ts)*1000),
        "success": bool(success),
        "fingerprint": fingerprint_input(input_payload),
        "ts": time.time()
    }
    try:
        queue.put_nowait(evt)
    except asyncio.QueueFull:
        global drop_counter
        drop_counter += 1
        # optionally increment a local dropped metric

# background task
async def sender_loop():
    os.makedirs(SPILL_DIR, exist_ok=True)
    session = aiohttp.ClientSession()
    spill_file = None
    while True:
        batch = []
        try:
            # gather up to BATCH_SIZE with timeout
            for _ in range(BATCH_SIZE):
                evt = await asyncio.wait_for(queue.get(), timeout=FLUSH_INTERVAL)
                batch.append(evt)
        except asyncio.TimeoutError:
            pass

        # if nothing to send, but there are spill files, try to send them
        if not batch:
            # attempt to resend spilled files
            for fname in sorted(os.listdir(SPILL_DIR)):
                path = os.path.join(SPILL_DIR, fname)
                with open(path, "r") as f:
                    payload = f.read()
                try:
                    async with session.post(AGG_URL, data=payload, timeout=5) as r:
                        if r.status == 200:
                            os.remove(path)
                        else:
                            break
                except Exception:
                    break
            await asyncio.sleep(0.1)
            continue

        # try send with retries/backoff
        payload = json.dumps(batch)
        backoff = 0.5
        sent = False
        for attempt in range(4):
            try:
                async with session.post(AGG_URL, data=payload, timeout=5) as r:
                    if 200 <= r.status < 300:
                        sent = True
                        break
            except Exception:
                pass
            await asyncio.sleep(backoff)
            backoff *= 2

        if not sent:
            # spill to disk (bounded by rotating files)
            fname = os.path.join(SPILL_DIR, f"spill_{int(time.time()*1000)}.ndjson")
            with open(fname, "w") as f:
                f.write("\n".join(json.dumps(e) for e in batch))

    await session.close()

# usage: start sender_loop in background event loop on container start

-- Step 1: daily per-user totals by event_date
WITH daily AS (
  SELECT
    user_id,
    DATE(occurred_at) AS event_date,
    SUM(amount) AS day_amount
  FROM transactions
  GROUP BY user_id, DATE(occurred_at)
),

-- Step 2: generate full calendar per user (optional to fill zeros)
users_dates AS (
  SELECT
    u.user_id,
    d::date AS event_date
  FROM (SELECT DISTINCT user_id FROM daily) u
  CROSS JOIN UNNEST(
    GENERATE_DATE_ARRAY(
      (SELECT MIN(event_date) FROM daily),
      (SELECT MAX(event_date) FROM daily)
    )
  ) AS d
)

SELECT
  ud.user_id,
  ud.event_date,
  COALESCE(d.day_amount, 0) AS day_amount,
  -- 30-day rolling sum: current day + previous 29 days
  SUM(COALESCE(d.day_amount, 0)) OVER (
    PARTITION BY ud.user_id
    ORDER BY ud.event_date
    RANGE BETWEEN INTERVAL '29' DAY PRECEDING AND CURRENT ROW
  ) AS rolling_30d_spend
FROM users_dates ud
LEFT JOIN daily d
  ON d.user_id = ud.user_id AND d.event_date = ud.event_date
ORDER BY ud.user_id, ud.event_date;