Define scoped Forgejo native backup on the Scaleway primary

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

scripts/activate_forgejo_primary_backup.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+"""Apply only the reviewed Forgejo backup contract and verify native completion."""
+import argparse
+from datetime import datetime,timezone
+import json
+from pathlib import Path
+import subprocess
+import time
+import yaml
+from state_hub_preflight_lane import assert_cluster
+ROOT=Path(__file__).resolve().parents[1]
+
+def run(k,receipt):
+    def cmd(args,payload=None):
+        r=subprocess.run(k+args,input=None if payload is None else json.dumps(payload).encode(),capture_output=True,timeout=60)
+        if r.returncode: raise RuntimeError('backup_activation_command_failed')
+        return r.stdout
+    def get(args): return json.loads(cmd(args))
+    assert_cluster(k)
+    desired=yaml.safe_load((ROOT/'helm/forgejo-db-cluster.yaml').read_text())['spec']['backup']
+    if desired['barmanObjectStore']['destinationPath']!='s3://railiance-platform-pg-backup/platform-pg/forgejo-db/': raise RuntimeError('wrong_destination')
+    live=get(['-n','databases','get','cluster','forgejo-db','-o','json'])
+    if live['spec'].get('backup') not in (None,desired): raise RuntimeError('existing_backup_differs')
+    ready=any(c['type']=='Ready' and c['status']=='True' for c in live['status']['conditions'])
+    if not ready: raise RuntimeError('source_not_ready')
+    for file in ('helm/forgejo-db-backup-networkpolicies.yaml','helm/forgejo-db-backup.yaml'):
+        cmd(['apply','--dry-run=server','-f',str(ROOT/file)])
+    patch=[{'op':'test','path':'/metadata/resourceVersion','value':live['metadata']['resourceVersion']},{'op':'add','path':'/spec/backup','value':desired}]
+    cmd(['-n','databases','patch','cluster','forgejo-db','--type=json','--dry-run=server','--patch-file=/dev/stdin'],patch)
+    cmd(['apply','-f',str(ROOT/'helm/forgejo-db-backup-networkpolicies.yaml')])
+    cmd(['-n','databases','patch','cluster','forgejo-db','--type=json','--patch-file=/dev/stdin'],patch)
+    receipt['backup_contract_applied']=True
+    receipt['started_at']=datetime.now(timezone.utc).isoformat()
+    cmd(['apply','-f',str(ROOT/'helm/forgejo-db-backup.yaml')])
+    for _ in range(120):
+        rows=get(['-n','databases','get','backups','-o','json'])['items']
+        rows=[r for r in rows if r.get('spec',{}).get('cluster',{}).get('name')=='forgejo-db']
+        good=[r for r in rows if r.get('status',{}).get('phase')=='completed']
+        if good:
+            chosen=max(good,key=lambda r:r['metadata']['creationTimestamp'])
+            receipt.update(backup_name=chosen['metadata']['name'],backup_id=chosen['status'].get('backupId'),backup_phase='completed')
+            break
+        time.sleep(5)
+    else: raise RuntimeError('fresh_backup_not_completed')
+    live=get(['-n','databases','get','cluster','forgejo-db','-o','json'])
+    checks={c['type']:c['status']=='True' for c in live['status']['conditions']}
+    receipt.update(production_ready=checks.get('Ready',False),continuous_archiving=checks.get('ContinuousArchiving',False),destination=desired['barmanObjectStore']['destinationPath'],retention='30d')
+    if not receipt['production_ready'] or not receipt['continuous_archiving']: raise RuntimeError('source_or_archiving_not_ready')
+    receipt['status']='verified'
+
+def main():
+    p=argparse.ArgumentParser(description=__doc__);p.add_argument('--kubeconfig',required=True);p.add_argument('--receipt',required=True);a=p.parse_args()
+    receipt={'schema':'platform.forgejo-primary-backup.v1','status':'failed'}
+    target=Path(a.receipt)
+    with target.open('x') as f:
+        try:run(['kubectl','--kubeconfig',a.kubeconfig],receipt)
+        except Exception:receipt['error']='activation_or_acceptance_failed'
+        finally:json.dump(receipt,f,indent=2)
+    return int(receipt['status']!='verified')
+if __name__=='__main__':raise SystemExit(main())