Podium Conversation History Export

Podium Conversation History Export

Overview

Bulk-export a Podium organization's historical conversations, reviews, and contacts into a corpus suitable for embedding into a vector store. This is the skill you run when the customer says "we have two years of knowledge in there" and the AI team wants every thread, every review, every contact note searchable by similarity. It is not a one-shot script — it is the full-export-plus-incremental-CDC pipeline that ingests the historical backlog once and then keeps the corpus current via nightly updated_at watermark passes.

The six production failures this skill prevents:

1. Cursor pagination drift — Podium's next_cursor is server-side state derived from a sort key plus a position. If a conversation is created, updated, or deleted mid-export, naive cursor walks duplicate records (an updated row reappears at the new position) or skip records (a row deleted between pages shifts the cursor's anchor). A correct walk pins the sort to a stable monotonic field and dedups on id.
2. Incremental CDC gaps via the updated_at watermark — naive updated_at > $watermark queries miss writes that happen at exactly the watermark second. Two writes within the same second on opposite sides of the boundary produce a permanent hole. Correct CDC uses >= with explicit overlap margin and dedups in the loader on (id, updated_at).
3. Attachment URL expiry mid-download — Podium attachment URLs are pre-signed S3-style URLs that expire on the order of 15 minutes. A bulk exporter that takes an hour will get 403 SignatureDoesNotMatch on every attachment whose URL was issued in the first quarter of the run. Correct downloaders detect the 403, fetch a fresh signed URL by attachment_id, and resume.
4. Oversized thread chunking failures — a 4000-message conversation thread (a long-running concierge thread for a high-touch RV dealer customer) blows the typical 8K-token embedding budget if naively concatenated. Chunking must be windowed with semantic boundaries (turn boundaries, day boundaries, idle gaps) and emit overlapping chunks for cross-window retrieval.
5. PII not redacted before embedding — vector stores are effectively eternal; once a customer's SSN, credit-card number, or address is embedded it cannot be unembedded without recomputing the index. Redact at chunk-emit time with the same PII pattern set used by podium-call-transcript-pipeline, before any vector is computed.
6. Export OOM on long threads — naively loading all messages of a 4000-message thread into memory before chunking blows the heap on the host running the export. Correct exports stream message-by-message into JSONL, then a separate pass streams JSONL into chunks. Memory cost stays O(window-size), not O(thread-size).

Prerequisites

• Python 3.10+
• podium-auth (this skill assumes a PodiumAuth instance is available; do not re-implement OAuth here)
• podium-rate-limit-survival (this skill assumes the calling layer obeys per-endpoint quotas — bulk export is the most rate-limit-aggressive workload in the pack)
• A persistent CDC watermark store — SQLite is the default; cdc_watermark.py ships with it
• Local disk for streaming JSONL output, gzip-compressed (typical 2-year org: ~1–10 GB raw, 200 MB–2 GB gzipped)
• An attachment-download target directory (S3 bucket, GCS bucket, or local path)
• The PII redaction pattern set from podium-call-transcript-pipeline (reused; do not fork)

Authentication

This skill does NOT implement OAuth. All HTTP calls flow through a podium_get() injected dependency that holds a PodiumAuth instance from the podium-auth skill — that layer handles token caching, 80%-TTL refresh, single-flight locks, and scope validation. Bootstrap by Read-ing your refresh-token file, instantiate PodiumAuth(client_id, client_secret, refresh_token), then pass the instance to every export script via --refresh-token-file and the env-var credential flags. The five scripts in this skill never construct credentials in-process; they delegate.

If you need to harden the auth path itself (rotation, decay monitoring, multi-tenant routing), Read podium-auth/SKILL.md and stack that skill on top — this skill is the bulk-data layer, not the auth layer.

Instructions

Step 1 → Step 6 below. Build in this order. Each step neutralizes one of the six production failure modes from the Overview, in the same order.

Step 1. Cursor-paginated full crawl (neutralizes cursor drift)

Pin the sort to a stable monotonic field (created_at ascending) and dedup on id in the loader. Persist the cursor after every successful page so a crash mid-walk resumes at the last successful page boundary, not at the start.

import asyncio, json, time
from pathlib import Path
from typing import AsyncIterator

CURSOR_PATH = Path("./.cursor.conversations.json")
PAGE_SIZE = 100   # Podium documents 100 as the max; do not exceed

async def crawl_conversations(podium_get, location_uid: str) -> AsyncIterator[dict]:
    """Yield conversations in created_at-ascending order. Resumable across crashes."""
    state = json.loads(CURSOR_PATH.read_text()) if CURSOR_PATH.exists() else {}
    cursor = state.get("cursor")
    seen_ids = set(state.get("seen_ids", []))   # bounded; trim periodically

    while True:
        params = {
            "location_uid": location_uid,
            "sort": "created_at:asc",
            "limit": PAGE_SIZE,
        }
        if cursor:
            params["cursor"] = cursor

        resp = await podium_get("/v4/conversations", params=params)
        body = resp.json()

        page = body.get("data", [])
        for row in page:
            if row["id"] in seen_ids:
                continue                 # dedup against mid-walk updates
            seen_ids.add(row["id"])
            yield row

        cursor = body.get("next_cursor")
        # Persist after EVERY page — crash resume must land on the last good cursor
        CURSOR_PATH.write_text(json.dumps({
            "cursor": cursor,
            "seen_ids": list(seen_ids)[-50_000:],   # keep last 50k ids only
            "updated_at": time.time(),
        }))

        if not cursor:
            return

The seen_ids set is bounded at 50k to prevent unbounded memory growth on a multi-million-row export. Tune the cap to roughly 5× PAGE_SIZE × pages_in_one_hour — large enough to dedup any reasonable update churn within the page-write window, small enough to fit in memory.

Step 2. Incremental CDC via overlap-margin watermark (neutralizes boundary gaps)

A naive updated_at > $watermark query misses any row whose updated_at is exactly the watermark second. Use >= and dedup in the loader; advance the watermark only after the page is fully persisted.

import sqlite3, time, json

WATERMARK_DB = "./watermarks.sqlite"

def get_watermark(resource: str) -> float:
    con = sqlite3.connect(WATERMARK_DB)
    cur = con.execute("SELECT watermark FROM cdc WHERE resource = ?", (resource,))
    row = cur.fetchone()
    con.close()
    return row[0] if row else 0.0

def advance_watermark(resource: str, new_watermark: float) -> None:
    con = sqlite3.connect(WATERMARK_DB)
    con.execute("""
        INSERT INTO cdc(resource, watermark, updated_at) VALUES(?, ?, ?)
        ON CONFLICT(resource) DO UPDATE SET watermark = excluded.watermark, updated_at = excluded.updated_at
    """, (resource, new_watermark, time.time()))
    con.commit()
    con.close()

async def incremental_pull(podium_get, resource: str, overlap_margin_s: int = 60):
    """Pull rows with updated_at >= (watermark - overlap_margin) and dedup."""
    watermark = get_watermark(resource)
    since = max(0, watermark - overlap_margin_s)        # explicit overlap

    cursor = None
    max_seen = watermark
    seen_keys = set()

    while True:
        params = {"updated_since": since, "sort": "updated_at:asc", "limit": 100}
        if cursor:
            params["cursor"] = cursor
        body = (await podium_get(f"/v4/{resource}", params=params)).json()

        for row in body.get("data", []):
            key = (row["id"], row["updated_at"])
            if key in seen_keys:
                continue
            seen_keys.add(key)
            yield row
            max_seen = max(max_seen, row["updated_at"])

        cursor = body.get("next_cursor")
        if not cursor:
            break

    # Advance watermark only after the full pass succeeds, not per-page —
    # a partial pass must re-pull from the previous watermark on retry.
    advance_watermark(resource, max_seen)

The overlap_margin_s = 60 is intentional. Podium's updated_at granularity is one second; a 60s overlap absorbs clock-skew and same-second writes without producing useful duplicate volume.