activity-core/src/activity_core/release_dispatch.py

"""Explicitly registered release dispatcher; no production credentials or admission."""
from __future__ import annotations

import asyncio
import re
from datetime import timedelta

from temporalio import activity, workflow
from temporalio.common import RetryPolicy
from temporalio.exceptions import ActivityError, ApplicationError

with workflow.unsafe.imports_passed_through():
    from activity_core.release_broker import TERMINAL, Broker


class ReleaseActivities:
    """Trusted worker construction binds the ledger, adapter factory and audit sink.

    The factory receives a ledger plan, never workflow-supplied configuration.
    The sink must durably upsert by event_id and return only after acknowledgement.
    This is at-least-once delivery: a lost acknowledgement can repeat an event.
    """

    def __init__(self, broker: Broker, backend_factory, sink):
        if not broker.admitted:
            raise ValueError("broker identity not admitted")
        self.broker, self.backend_factory, self.sink = broker, backend_factory, sink
        with broker.connect() as db:
            db.execute("CREATE TABLE IF NOT EXISTS release_audit_delivered "
                       "(sequence INTEGER PRIMARY KEY)")

    def _plan(self, release_id):
        import json

        if not self.broker.admitted:
            raise ApplicationError("release identity disabled", non_retryable=True)
        if not isinstance(release_id, str) or not re.fullmatch(r"[0-9a-f]{64}", release_id):
            raise ApplicationError("invalid release id", non_retryable=True)
        with self.broker.connect() as db:
            row = db.execute("SELECT plan FROM releases WHERE id=?", (release_id,)).fetchone()
        if row is None:
            raise ApplicationError("unknown admitted release", non_retryable=True)
        return json.loads(row[0])

    def _advance(self, release_id, plan):
        return self.broker.advance(release_id, self.backend_factory(plan))

    def _deliver(self, release_id, plan):
        with self.broker.connect() as db:
            rows = db.execute(
                "SELECT sequence, phase, observed_at FROM transitions "
                "WHERE release_id=? AND sequence NOT IN "
                "(SELECT sequence FROM release_audit_delivered) "
                "ORDER BY sequence LIMIT 100", (release_id,),
            ).fetchall()
        for sequence, phase, observed_at in rows:
            # Only normalized facts leave the ledger. No envelopes or adapter errors.
            self.sink({
                "event_id": f"release:{release_id}:{sequence}",
                "event_type": "release_transition", "release_id": release_id,
                "application": "activity-core", "phase": phase,
                "observed_at": observed_at, "candidate": plan["candidate"],
                "rollback": plan["rollback"],
            })
            with self.broker.connect() as db:
                db.execute("INSERT OR IGNORE INTO release_audit_delivered VALUES (?)",
                           (sequence,))
        return len(rows)

    @activity.defn(name="advance_admitted_release")
    async def advance(self, release_id: str) -> str:
        plan = self._plan(release_id)
        try:
            return await asyncio.to_thread(self._advance, release_id, plan)
        except Exception:
            # Do not put transport exception text or credentials in Temporal history.
            raise ApplicationError("release step unavailable", type="ReleaseStepUnavailable") from None

    @activity.defn(name="deliver_release_audit")
    async def deliver(self, release_id: str) -> int:
        plan = self._plan(release_id)
        try:
            return await asyncio.to_thread(self._deliver, release_id, plan)
        except Exception:
            raise ApplicationError("release audit unavailable", type="ReleaseAuditUnavailable") from None


@workflow.defn
class ReleaseDispatchWorkflow:
    """Resume one admitted ledger entry; receipt admission stays outside Temporal."""

    @workflow.run
    async def run(self, release_id: str) -> str:
        for _ in range(100):
            phase = await workflow.execute_activity(
                "advance_admitted_release", release_id,
                start_to_close_timeout=timedelta(minutes=10),
                retry_policy=RetryPolicy(maximum_interval=timedelta(seconds=30)),
            )
            try:
                await workflow.execute_activity(
                    "deliver_release_audit", release_id,
                    start_to_close_timeout=timedelta(minutes=1),
                    retry_policy=RetryPolicy(
                        maximum_interval=timedelta(seconds=30),
                        maximum_attempts=0 if phase in TERMINAL else 1,
                    ),
                )
            except ActivityError:
                if phase in TERMINAL:
                    raise
                # A sink outage must not prevent rollback. Committed transitions
                # stay pending and terminal completion waits for audit delivery.
                workflow.logger.warning("Release audit pending; ledger retains transitions")
            if phase in TERMINAL:
                return phase
            if phase == "rollback_synced":
                await workflow.sleep(timedelta(seconds=30))
        workflow.continue_as_new(release_id)


def release_worker(client, activities: ReleaseActivities):
    """Dedicated queue: never add privileged release activities to orchestrator-tq.

    Call only from an admitted deployment with durable shared ledger storage,
    immutable transport configuration and a bounded, idempotent audit sink.
    This helper starts no worker, workflow or schedule by itself.
    """
    from temporalio.worker import Worker

    if not activities.broker.admitted:
        raise ValueError("broker identity not admitted")
    return Worker(client, task_queue="activity-core-release-tq",
                  workflows=[ReleaseDispatchWorkflow],
                  activities=[activities.advance, activities.deliver])