erp/.claude/skills/dolibarr-sandbox-write/scripts/promote-apply.sh

#!/usr/bin/env bash
# Replay a promote manifest against a target — sandbox (rehearsal) or prod (real).
#
# Resolves the manifest's symbolic @refs to the ids actually created during this
# run, so dependent ops (an invoice -> its just-created thirdparty) wire up on the
# target. Sandbox uses dol-write.sh; prod uses the gated dol-prod-write.sh.
#
#   promote-apply.sh <manifest.json> [--target sandbox|prod]
#
# --target prod requires, in the environment (never stored):
#   DOLIBARR_PROD_WRITE_KEY=<prod write key>
#   ARCO_PROMOTE_CONFIRM=I-UNDERSTAND-THIS-WRITES-PROD
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
MANIFEST="${1:?usage: promote-apply.sh <manifest.json> [--target sandbox|prod]}"; shift || true
TARGET="sandbox"
while [[ $# -gt 0 ]]; do
  case "$1" in
    --target) TARGET="${2:?}"; shift 2 ;;
    *) echo "promote-apply.sh: unknown arg '$1'" >&2; exit 2 ;;
  esac
done
case "${TARGET}" in
  # sandbox honors a pre-set DOL_WRITE so the offline tests can inject
  # tests/stub-dol-write.sh (the erp#37 hook the op scripts already honor);
  # the default is the host-guarded dol-write.sh.
  sandbox) export DOL_WRITE="${DOL_WRITE:-${SCRIPT_DIR}/dol-write.sh}" ;;
  # prod NEVER inherits — always the gated dol-prod-write.sh (env-only key
  # + ARCO_PROMOTE_CONFIRM), so no environment trick can reroute a prod apply.
  prod)    export DOL_WRITE="${SCRIPT_DIR}/dol-prod-write.sh" ;;
  *) echo "promote-apply.sh: --target must be sandbox|prod" >&2; exit 2 ;;
esac
echo ">>> promote-apply target=${TARGET} (writes via $(basename "${DOL_WRITE}"))" >&2

python3 - "$MANIFEST" "$SCRIPT_DIR" <<'PY'
import json, sys, subprocess, os
manifest_path, script_dir = sys.argv[1], sys.argv[2]
ops = json.load(open(manifest_path))
OP_SCRIPT = {"thirdparty": "thirdparty-create.sh", "invoice": "invoice-create.sh",
             "creditnote": "creditnote-create.sh", "payment": "payment-record.sh",
             "thirdparty_update": "thirdparty-update.sh", "contact": "contact-create.sh",
             "attach": "document-attach.sh"}
refmap = {}
n_created = n_deduped = 0

import urllib.parse
DOL_WRITE = os.environ.get("DOL_WRITE")  # GET wrapper for the chosen target

# Business-key lookup of a PRE-EXISTING entity on the target, so a manifest can
# reference records it does not create (e.g. an invoice for an existing client).
# Resolves against the *target*, so #thirdparty:name=X becomes the sandbox id on
# --target sandbox and the prod id on --target prod (portability for real changes).
ENTITY_LOOKUP = {
    "thirdparty": ("/thirdparties", {"name": "t.nom", "code": "t.code_client",
                                     "supplier_code": "t.code_fournisseur"}),
    "invoice": ("/invoices", {"ref": "t.ref"}),
    "supplierinvoice": ("/supplierinvoices", {"ref": "t.ref", "ref_supplier": "t.ref_supplier"}),
}
def lookup(spec):
    try:
        ent, rest = spec.split(":", 1); field, value = rest.split("=", 1)
    except ValueError:
        sys.exit("promote-apply: bad lookup '#%s' (use #entity:field=value)" % spec)
    if ent not in ENTITY_LOOKUP:
        sys.exit("promote-apply: unknown lookup entity '%s'" % ent)
    endpoint, cols = ENTITY_LOOKUP[ent]
    if field not in cols:
        sys.exit("promote-apply: cannot look %s up by '%s' (try %s)" % (ent, field, "/".join(cols)))
    flt = "(%s:=:'%s')" % (cols[field], value.replace("'", "''"))
    path = "%s?limit=2&sqlfilters=%s" % (endpoint, urllib.parse.quote(flt))
    r = subprocess.run([DOL_WRITE, "GET", path], capture_output=True, text=True, env=os.environ)
    if r.returncode != 0:
        sys.stderr.write(r.stdout + r.stderr + "\n")
        sys.exit("promote-apply: lookup '#%s' query failed" % spec)
    try:
        rows = json.loads(r.stdout)
    except Exception:
        rows = []
    rows = rows if isinstance(rows, list) else []
    if len(rows) == 0:
        sys.exit("promote-apply: lookup '#%s' matched nothing on the target" % spec)
    if len(rows) > 1:
        sys.exit("promote-apply: lookup '#%s' is ambiguous (%d matches) — use a unique key" % (spec, len(rows)))
    return int(rows[0]["id"])

def resolve(v):
    if isinstance(v, str) and v.startswith("@"):
        k = v[1:]
        if k not in refmap:
            sys.exit("promote-apply: unresolved ref @%s (is it created earlier in the manifest?)" % k)
        return refmap[k]
    if isinstance(v, str) and v.startswith("#"):
        return lookup(v[1:])
    if isinstance(v, dict):
        return {kk: resolve(vv) for kk, vv in v.items()}
    if isinstance(v, list):
        return [resolve(x) for x in v]
    return v
for i, op in enumerate(ops, 1):
    t = op["op"]; script = OP_SCRIPT.get(t)
    if not script:
        sys.exit("promote-apply: unknown op '%s'" % t)
    inp = resolve(op.get("input", {}))
    # attach: a relative "file" is relative to the MANIFEST, not the CWD — the
    # manifest is the portable unit (replay packs carry their pdfs/ beside it).
    if t == "attach" and isinstance(inp.get("file"), str) and not os.path.isabs(inp["file"]):
        inp["file"] = os.path.normpath(os.path.join(
            os.path.dirname(os.path.abspath(manifest_path)), inp["file"]))
    r = subprocess.run([os.path.join(script_dir, script)], input=json.dumps(inp),
                       capture_output=True, text=True, env=os.environ)
    if r.returncode != 0:
        sys.stderr.write(r.stdout + r.stderr + "\n")
        sys.exit("promote-apply: op %d (%s) FAILED" % (i, t))
    out = r.stdout.strip()
    parsed = None
    try:
        parsed = json.loads(out)
    except Exception:
        pass
    if isinstance(parsed, dict):
        rid = parsed.get("id")
    else:
        rid = out if out.isdigit() else None
    ref = op.get("ref")
    if ref and rid is not None:
        refmap[ref] = int(rid) if str(rid).isdigit() else rid
    # Surface the idempotency evidence inline: every op script reports
    # deduped=true/false (erp#44), and thirdparty-update reports its read-back
    # diff — together the proof a re-apply is a no-op.
    extra = ""
    if isinstance(parsed, dict):
        if parsed.get("deduped"):
            n_deduped += 1
            extra += "  deduped=true (already on target — no write)"
        elif "deduped" in parsed:
            n_created += 1
            extra += "  created"
        ch = parsed.get("changed")
        if isinstance(ch, dict):
            extra += "  changed=%d%s" % (len(ch),
                     (" [%s]" % ", ".join(sorted(ch))) if ch else " (no-op)")
    print("  [%d/%d] %-17s %-8s -> id=%s%s" % (i, len(ops), t, ("@" + ref) if ref else "",
                                               rid if rid is not None else "-", extra))
counts = []
if n_created:
    counts.append("%d created" % n_created)
if n_deduped:
    counts.append("%d deduped" % n_deduped)
print("OK — promote complete%s. ref -> id: %s"
      % ((" (%s)" % ", ".join(counts)) if counts else "", json.dumps(refmap)))
PY