proofpoint-quarantine-portal/pipeline.py

"""Configurable Report & Release pipeline.

"Report & Release" is a sequence of steps run per message chunk, chosen in the admin
panel by checkboxes. The order is FIXED: release → move → delete. Config only toggles
which steps are included (a subsequence of ALLOWED_STEPS).

Why the order is load-bearing (see AGENTS.md):

* `release` MUST run first and MUST NOT pass `deletedfolder`. Releasing in place leaves
  the message in its folder so its `localguid` stays valid for a following `move`.
  Passing `deletedfolder` would relocate the message and invalidate every localguid —
  the exact bug the guid-refind fallback exists to paper over.
* `move`/`delete` relocate the message, so afterwards the localguids are stale. A step
  returns `localguids=None` to signal that; the runner re-finds them by the stable
  `guid` — but LAZILY, only when another step actually follows. So the common
  `["release", "move"]` still issues exactly two PPS calls, identical to the PoC.

Each step acts on the message's CURRENT location, threading (folder, localguids)
forward. Adding a future action = one `@step(...)` function + an entry in ALLOWED_STEPS;
no worker or frontend change.
"""

from __future__ import annotations

import logging
import time
from dataclasses import dataclass, replace
from typing import Any, Callable, Mapping

from pps_client import PPSError

log = logging.getLogger("pps.pipeline")

# Canonical order. Also the validation whitelist (config_store imports this).
ALLOWED_STEPS: tuple[str, ...] = ("release", "move", "delete")

# Seconds to wait between two steps on the same message when config omits the setting.
# PPS's backend is eventually-consistent: acting again too soon (e.g. move right after
# release) can hit a stale view. The delay lets the backend catch up.
DEFAULT_STEP_DELAY = 60


@dataclass(frozen=True)
class StepContext:
    pps: Any
    cfg: Mapping[str, Any]           # the quarantine config section
    folder: str                      # where the messages are RIGHT NOW
    localguids: list[str] | None     # None => stale; runner must re-find by guid
    chunk: list[dict]                # job_items rows: localguid, guid, subject, ...


@dataclass(frozen=True)
class StepResult:
    folder: str
    localguids: list[str] | None
    dst: str | None                  # folder this step lands messages in (for ops log)


StepFn = Callable[[StepContext], StepResult]
STEPS: dict[str, StepFn] = {}


def step(name: str):
    def register(fn: StepFn) -> StepFn:
        STEPS[name] = fn
        return fn
    return register


@step("release")
def _release(ctx: StepContext) -> StepResult:
    # NO deletedfolder — this is the localguid-stability rule. Never add it here.
    ctx.pps.act("release", ctx.folder, ctx.localguids)
    return StepResult(folder=ctx.folder, localguids=ctx.localguids, dst=None)


@step("move")
def _move(ctx: StepContext) -> StepResult:
    target = ctx.cfg["report_release"]["move_target"]
    if not target:
        raise ValueError("report_release 'move' step requires move_target")
    try:
        ctx.pps.act("move", ctx.folder, ctx.localguids, targetfolder=target)
    except PPSError:
        # Some deployments' release relocates the message to the deleted folder (like the
        # admin GUI). The move from `folder` then fails; re-find by stable guid and move
        # from the deleted folder instead.
        deleted = ctx.cfg.get("deleted_folder")
        if not deleted:
            raise PPSError("release relocated messages but no deleted_folder configured")
        found = refind_localguids(ctx.pps, ctx.cfg, deleted, ctx.chunk)
        ctx.pps.act("move", deleted, found, targetfolder=target)
    return StepResult(folder=target, localguids=None, dst=target)


@step("delete")
def _delete(ctx: StepContext) -> StepResult:
    deleted = ctx.cfg.get("deleted_folder")
    ctx.pps.act("delete", ctx.folder, ctx.localguids, deletedfolder=deleted)
    return StepResult(folder=deleted, localguids=None, dst=deleted)


def run_pipeline(pps, cfg: Mapping[str, Any], folder: str, chunk: list[dict],
                 sleep: Callable[[float], None] = time.sleep) -> None:
    """Execute the configured steps in order for one chunk of messages.

    Between two steps we wait `report_release.step_delay_seconds` (default 60) so PPS's
    eventually-consistent backend settles before the next action on the same message —
    without this, e.g. a move right after a release can act on a stale view. The wait
    blocks the worker thread; it does NOT block the HTTP request (actions are queued and
    return 202 immediately). `sleep` is injectable for tests.
    """
    steps = cfg["report_release"]["steps"]
    delay = int(cfg["report_release"].get("step_delay_seconds", DEFAULT_STEP_DELAY))
    ctx = StepContext(
        pps=pps,
        cfg=cfg,
        folder=folder,
        localguids=[it["localguid"] for it in chunk],
        chunk=chunk,
    )
    for i, name in enumerate(steps):
        # Delay before every step except the first — also lets the backend settle before
        # the refind search below finds a just-relocated message.
        if i > 0 and delay > 0:
            log.info("waiting %ds before %r step (PPS backend consistency)", delay, name)
            sleep(delay)
        if ctx.localguids is None:  # a prior relocating step left them stale
            ctx = replace(ctx, localguids=refind_localguids(pps, cfg, ctx.folder, chunk))
        result = STEPS[name](ctx)
        ctx = replace(ctx, folder=result.folder, localguids=result.localguids)


def refind_localguids(pps, cfg: Mapping[str, Any], folder: str, chunk: list[dict]) -> list[str]:
    """Recover current localguids in `folder` by matching the stable guid.

    Generalises the PoC's _move_from_deleted_by_guid. Raises PPSError if nothing matches.
    """
    wanted = {it["guid"]: it for it in chunk if it.get("guid")}
    if not wanted:
        raise PPSError("cannot recover messages: no guids stored for this chunk")
    records = pps.search(
        folder,
        cfg.get("list_query", "from=*"),
        limit=1000,
        days_back=int(cfg.get("default_days_back", 7)),
    )
    found = [
        r["localguid"] for r in records
        if r.get("guid") in wanted and r.get("localguid")
    ]
    if not found:
        raise PPSError(f"messages not found in {folder!r} to continue the pipeline")
    return found


def pipeline_destination(cfg: Mapping[str, Any]) -> str | None:
    """Final folder after all steps, computed WITHOUT touching PPS (for the ops log)."""
    folder = None
    for name in cfg["report_release"]["steps"]:
        if name == "move":
            folder = cfg["report_release"]["move_target"]
        elif name == "delete":
            folder = cfg.get("deleted_folder")
    return folder


def describe_pipeline(cfg: Mapping[str, Any]) -> str:
    """Human sentence for the admin panel, e.g.
    'released in place → moved to "Debugging - Josef" → deleted to "Deleted"'."""
    rr = cfg.get("report_release", {})
    phrases = []
    for name in rr.get("steps", []):
        if name == "release":
            phrases.append("released in place")
        elif name == "move":
            phrases.append(f'moved to "{rr.get("move_target", "?")}"')
        elif name == "delete":
            phrases.append(f'deleted to "{cfg.get("deleted_folder", "?")}"')
    return " → ".join(phrases) if phrases else "(no steps configured)"