main.py

"""Render Workflows app: task registration and the `run_sync` graph.

This file only registers tasks and wires them together. Every implementation is a
plain function under `tasks/` or `connectors/`, which keeps the SDK out of the
unit tests and keeps this file readable as a description of the pipeline.

Task plans, timeouts, and retries are set per task below. They are not incidental
— see the comments; several of them are the difference between a correct sync and
one that quietly duplicates or loses data.
"""

from __future__ import annotations

import asyncio
import logging

from render_sdk import Retry, Workflows

from connectors.base import load_connector
from db import log_event
from tasks.backfill import backfill_window
from tasks.backfill import fetch_backfill_slice as _fetch_backfill_slice
from tasks.backfill import open_backfill as _open_backfill
from tasks.fetch import fetch_slice as _fetch_slice
from tasks.load import load_batch as _load_batch
from tasks.migrate import migrate as _migrate
from tasks.state import advance_watermark as _advance_watermark
from tasks.state import complete_without_watermark as _complete_without_watermark
from tasks.state import fail_run, new_run_id, open_run as _open_run

app = Workflows(
    default_retry=Retry(max_retries=2, wait_duration_ms=2000, backoff_scaling=2),
    default_timeout=900,
    default_plan="standard",
)


# `open_run` and `advance_watermark` are small enough to be plain function calls,
# but they are registered tasks anyway so each one shows up as its own row in the
# Dashboard run tree. Being able to see where a run got to is a large part of what
# this template is for.
@app.task(
    plan="starter",
    timeout_seconds=120,
    retry=Retry(max_retries=3, wait_duration_ms=2000, backoff_scaling=2),
)
def open_run(run_id: str, connector: str | None = None) -> dict:
    """Claim a run row and compute the window this run is responsible for."""
    return _open_run(load_connector(connector), run_id=run_id)


@app.task(
    plan="starter",
    timeout_seconds=300,
    retry=Retry(max_retries=3, wait_duration_ms=2000, backoff_scaling=2),
)
def migrate(connector: str | None = None) -> dict:
    return _migrate(load_connector(connector))


# `starter` because this task is I/O-bound waiting on the source API, not on CPU.
# Per-task instance sizing is a Workflows feature, and this is a case where the
# smaller plan is also the correct choice: 8 of these run concurrently.
@app.task(
    plan="starter",
    timeout_seconds=1800,
    retry=Retry(max_retries=3, wait_duration_ms=2000, backoff_scaling=2),
)
def fetch_slice(run_id: str, slice_index: int, connector: str | None = None) -> dict:
    """Paginate one time window into staging, checkpointing after every page.

    Safe to retry: it resumes from `sync_slice.cursor` rather than restarting the
    window, so a crash on page 47 costs page 47 and nothing else.
    """
    return _fetch_slice(load_connector(connector), run_id, slice_index)


@app.task(
    plan="standard",
    timeout_seconds=900,
    retry=Retry(max_retries=2, wait_duration_ms=5000, backoff_scaling=2),
)
def load_batch(run_id: str, connector: str | None = None) -> dict:
    """Staging → target, as one idempotent guarded upsert."""
    return _load_batch(load_connector(connector), run_id)


@app.task(
    plan="starter",
    timeout_seconds=60,
    retry=Retry(max_retries=3, wait_duration_ms=2000, backoff_scaling=2),
)
def advance_watermark(run_id: str) -> dict:
    """Move the watermark forward. Refuses unless `load_batch` committed."""
    return _advance_watermark(run_id)


# Retries are disabled deliberately. Retrying an orchestrator re-executes it from
# the top with no memory of the prior attempt's children, which turns a timeout
# into a silent duplicate fan-out. A failure here should be terminal and visible
# to the cron trigger instead. This is a non-obvious call that readers will
# otherwise "fix" — please don't.
@app.task(
    plan="standard",
    timeout_seconds=3600,
    retry=Retry(max_retries=0, wait_duration_ms=0),
)
async def run_sync(connector: str | None = None, **_unused_input) -> dict:
    """Orchestrator: migrate → open_run → fetch_slice × N → load_batch → advance_watermark.

    The ordering is the contract. The watermark only moves after the load commits,
    so any failure anywhere above it leaves the window un-acknowledged and the
    next run covers it again.
    """
    # Resolve once and pass the explicit name to every fresh task instance. An
    # environment change during a long run cannot switch connectors mid-graph.
    connector = connector or load_connector().name
    summary: dict = {
        "status": "running",
        "connector": connector,
        "unused_input_keys": sorted(_unused_input.keys()),
    }
    run_id = new_run_id()

    try:
        await migrate(connector=connector)

        run = await open_run(run_id=run_id, connector=connector)
        summary["run"] = run

        slice_indexes = run["slice_indexes"]
        if not slice_indexes:
            # A clock rollback or a watermark exactly equal to `now` can produce
            # a no-op. open_run has already committed and released its lease.
            summary["status"] = "empty"
            log_event(logging.INFO, "sync_empty_window", run_id=run_id)
            return summary

        # Slices are independent by construction, so this is a genuine fan-out.
        # Each one is its own task run and appears as its own node in the run tree.
        slices = await asyncio.gather(
            *(
                fetch_slice(run_id=run_id, slice_index=index, connector=connector)
                for index in slice_indexes
            )
        )
        summary["slices"] = slices
        summary["rows_staged"] = sum(result["rows_staged"] for result in slices)

        summary["load"] = await load_batch(run_id=run_id, connector=connector)
        summary["watermark"] = await advance_watermark(run_id=run_id)
        summary["status"] = "completed"

        log_event(
            logging.INFO,
            "sync_completed",
            run_id=run_id,
            rows_staged=summary["rows_staged"],
            rows_upserted=summary["load"]["rows_upserted"],
            watermark=summary["watermark"]["watermark"],
        )
        return summary
    except BaseException as exc:
        # BaseException so a timeout or cancellation still records a failed run
        # rather than leaving a row stuck in `running` forever.
        summary["status"] = "failed"
        summary["error"] = str(exc) or exc.__class__.__name__
        try:
            fail_run(run_id)
        except Exception as mark_exc:
            log_event(
                logging.WARNING,
                "run_status_update_failed",
                run_id=run_id,
                error=str(mark_exc),
            )
        log_event(logging.ERROR, "sync_failed", run_id=run_id, error=summary["error"])
        raise


@app.task(
    plan="starter",
    timeout_seconds=120,
    retry=Retry(max_retries=3, wait_duration_ms=2000, backoff_scaling=2),
)
def open_backfill(
    run_id: str,
    since: str,
    until: str,
    connector: str | None = None,
) -> dict:
    return _open_backfill(
        load_connector(connector),
        run_id=run_id,
        since=since,
        until=until,
    )


@app.task(
    plan="starter",
    timeout_seconds=6000,
    retry=Retry(max_retries=3, wait_duration_ms=2000, backoff_scaling=2),
)
def fetch_backfill_slice(run_id: str, connector: str | None = None) -> dict:
    """Page historical objects with the same durable checkpoint loop as sync."""
    return _fetch_backfill_slice(load_connector(connector), run_id)


@app.task(
    plan="starter",
    timeout_seconds=60,
    retry=Retry(max_retries=3, wait_duration_ms=2000, backoff_scaling=2),
)
def complete_backfill(run_id: str) -> dict:
    return _complete_without_watermark(run_id)


# Stripe retains events for 30 days, so older history has to page the object
# endpoint by `created`. This is a separate no-retry orchestrator for the same
# reason as run_sync: only its child page task retries, using the persisted
# cursor. Every phase remains visible in the Dashboard run tree.
@app.task(
    plan="standard",
    # Four possible 6,000-second fetch attempts plus load/finalization headroom.
    timeout_seconds=28800,
    retry=Retry(max_retries=0, wait_duration_ms=0),
)
async def backfill(
    since: str,
    until: str | None = None,
    connector: str | None = None,
) -> dict:
    connector = connector or load_connector().name
    window_start, window_end = backfill_window(since, until)
    run_id = new_run_id("backfill")
    summary: dict = {
        "run_id": run_id,
        "connector": connector,
        "window_start": window_start.isoformat(),
        "window_end": window_end.isoformat(),
        "status": "running",
    }

    try:
        await migrate(connector=connector)
        summary["run"] = await open_backfill(
            run_id=run_id,
            since=window_start.isoformat(),
            until=window_end.isoformat(),
            connector=connector,
        )
        summary["fetch"] = await fetch_backfill_slice(
            run_id=run_id,
            connector=connector,
        )
        summary["load"] = await load_batch(run_id=run_id, connector=connector)
        summary["completion"] = await complete_backfill(run_id=run_id)
        summary["status"] = "completed"
        return summary
    except BaseException:
        try:
            fail_run(run_id)
        except Exception as mark_exc:
            log_event(
                logging.WARNING,
                "backfill_status_update_failed",
                run_id=run_id,
                error=str(mark_exc),
            )
        raise


if __name__ == "__main__":
    app.start()