From c9984e7ae6fbdcbdd7d447bbe835535c53c928ba Mon Sep 17 00:00:00 2001 From: NguyenND Date: Tue, 11 Aug 2026 11:30:57 -0400 Subject: [PATCH] Aug 11 - Update photo storage using r2 --- CLAUDE.md | 71 +++++- MULTI_TENANT_PLAN.md | 55 +++++ requirements.txt | 1 + scripts/audit_photos.py | 390 ++++++++++++++++++++++++++++++++ scripts/migrate_photos_to_r2.py | 329 +++++++++++++++++++++++++++ tests/test_storage_backend.py | 154 +++++++++++++ 6 files changed, 999 insertions(+), 1 deletion(-) create mode 100644 scripts/audit_photos.py create mode 100644 scripts/migrate_photos_to_r2.py create mode 100644 tests/test_storage_backend.py diff --git a/CLAUDE.md b/CLAUDE.md index 554556b..273ffac 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -34,6 +34,7 @@ 24. [Backup CLI](#24-backup-cli-controlbackuppy) 25. [Health Dashboard](#25-health-dashboard-health-on-panel) 26. [Coding Rules for AI Assistants](#26-coding-rules-for-ai-assistants) +29. [Photo Object Storage (R2)](#29-photo-object-storage-r2) --- @@ -1937,4 +1938,72 @@ curl -sI -H "Host: ztest.jqc.app" http://127.0.0.1:8000/ | head -2 # 5. Wildcard TLS cert (DNS-01, certbot) # 6. Add to /etc/jqc/control.env: MULTI_TENANT_ENABLED=true # 7. sudo systemctl daemon-reload && sudo systemctl restart jqc -``` \ No newline at end of file +``` +--- + +## 29. Photo Object Storage (R2) + +**Goal:** photo **files** live in Cloudflare R2 (S3-compatible, $0 egress), not on +the server's local disk. The DB is not the bottleneck — rows are tiny, PDFs are +streamed via `BytesIO` and never written to disk. Only photos accumulate, and in +a multi-tenant deployment they accumulate from every tenant onto one volume. + +**No schema change, ever.** The DB stores an unprefixed relative path +(`uploads/issue_photos/abc.jpg`) and continues to. That string is the storage +**key**; `app/utils/storage.py` maps it to a backend. + +### Key mapping (the rule that matters) + +| Layer | Value | +|---|---| +| DB (`issues.photo_path`, `result_photos[]`, `mobile_photo_paths[]`, `inspections.form_data`, `inspection_results.photo_path`) | `uploads/issue_photos/abc.jpg` | +| Local backend, on disk | `app/static/uploads/issue_photos/abc.jpg` | +| S3 backend, object key | `t/uploads/issue_photos/abc.jpg` | + +The `t/` prefix is applied **only** inside `S3Backend._object_key()`, +from `g.tenant`. It never enters the DB, a template, or an API payload — so the +tenant DB stays portable and every caller stays tenant-agnostic. The **local** +backend deliberately does not prefix: prefixing would relocate every existing +file, and the local layout must remain byte-identical to what predates the seam. +Local mode therefore shares one uploads directory across tenants — an isolation +weakness inherited from before multi-tenancy, and the reason to move to `s3`. + +`S3Backend.save()` **raises** when `MULTI_TENANT_ENABLED` is true and no tenant +is bound, rather than writing an unprefixed key that a second tenant could later +collide with. Reads are more forgiving: `read()` / `exists()` / `delete()` try +the prefixed key and then the bare key, so objects written before the prefix +existed stay reachable. Pinned by `tests/test_storage_backend.py`. + +### Config (all env, per deployment) + +`STORAGE_BACKEND=local|s3` (default `local`), plus `R2_ENDPOINT_URL`, +`R2_ACCESS_KEY_ID`, `R2_SECRET_ACCESS_KEY`, `R2_BUCKET`, `R2_PRESIGN_TTL` +(default 86400), `R2_MEDIA_FALLBACK` (default false). `boto3` is imported lazily, +so `local` deployments never touch it — but it **is** in `requirements.txt`, +because `STORAGE_BACKEND=s3` fails at first upload without it. + +`STORAGE_BACKEND` is process-wide, not per-tenant: one bucket, one backend, all +tenants, isolated by prefix. A per-tenant backend would need the resolver to +carry a storage selector and `get_backend()` to cache per tenant instead of per +app — do not half-build it. + +**CSP:** `set_security_headers` in `app/__init__.py` derives the R2 host from +`R2_ENDPOINT_URL` and appends it to `img-src` automatically. Presigned images are +blocked by the browser without this. A custom R2 domain must be added too. + +### Operator scripts + +- `scripts/audit_photos.py` — read-only. Per tenant, collects every key its DB + references and reconciles against disk. Records the baseline that must still + resolve after cutover, and flags any key claimed by more than one tenant. +- `scripts/migrate_photos_to_r2.py` — copy-only, idempotent, resumable, + MD5+size verified. Uploads each tenant's referenced files to `t/…`. + Exit 0 only when everything verifies. Orphans (referenced by no tenant) are + **not** uploaded — no prefix could legitimately claim them. + +Both establish file ownership from tenant DB references, because the shared +local directory carries none. Both need `CONTROL_DATABASE_URL` + +`CONTROL_FERNET_KEY`; neither writes to any database. + +**Rollback is one env var:** `STORAGE_BACKEND=local` + restart. The sync never +deletes local files, so the old tree is intact indefinitely. diff --git a/MULTI_TENANT_PLAN.md b/MULTI_TENANT_PLAN.md index 78c61aa..cdd1588 100644 --- a/MULTI_TENANT_PLAN.md +++ b/MULTI_TENANT_PLAN.md @@ -227,6 +227,61 @@ Stripe per-plan subscription, fully implemented in `app/billing/` (`routes.py`, **MT-9 — iOS multi-tenant. 🔲 PENDING (server + client both unbuilt).** Planned server side: `GET /api/v1/discover?subdomain=acme` and `GET /api/v1/tenant` public endpoints — to be exempt from tenant middleware via `MULTI_TENANT_EXEMPT_PATHS`. **Neither endpoint exists in the code yet** — no `api_discovery` blueprint is registered. iOS side: also pending (web-first priority). +**MT-20 — R2 photo storage cutover. ✅ CODE DONE / 🔲 CUTOVER PENDING.** +No schema change, no migration. The data-plane code was already fully on the +storage seam (`storage.save()` / `media_url()` / `materialize_to_dir()` / +`delete()` at every inspection and issue photo site) — MT served local purely +because `STORAGE_BACKEND` had never been flipped, and it **could not** be: + +- `boto3` was missing from `requirements.txt`, so `S3Backend.__init__`'s lazy + import would have raised `ModuleNotFoundError` on the first upload after the + flip — at request time, not at boot. Now pinned (`boto3>=1.34`, matching ST). +- there was no cutover tooling, and ST's could not be reused (below). + +**Why ST's sync script does not port.** The local backend deliberately does not +prefix keys, so one shared `app/static/uploads/` holds every tenant's photos and +a file on disk carries no ownership marker. ST's script walks the disk and +uploads everything, which on MT writes objects with no tenant prefix — keys +`S3Backend._object_key()` will never read. Ownership must instead be derived +from each tenant DB's references, then written under `t/`. + +| Layer | Value | +|---|---| +| DB | `uploads/issue_photos/abc.jpg` | +| Disk (local backend) | `app/static/uploads/issue_photos/abc.jpg` | +| R2 object (s3 backend) | `t3/uploads/issue_photos/abc.jpg` | + +Delivered: +- `scripts/audit_photos.py` — read-only. Per tenant, collects every key its DB + references (5 sources: `issues.photo_path`, `.mobile_photo_paths[]`, + `.result_photos[]`, `inspections.form_data`, `inspection_results.photo_path`) + and reconciles against disk. Emits the baseline count that must still resolve + after cutover, plus orphans and any key claimed by more than one tenant. +- `scripts/migrate_photos_to_r2.py` — copy-only, idempotent, resumable, + MD5+size verified, tenant-prefixed. Exit 0 only on full verification. + Orphans are **not** uploaded: no prefix could legitimately claim them. +- `tests/test_storage_backend.py` — pins the prefix arithmetic, the bare key + returned to the DB, the refusal to write unprefixed with no tenant bound, and + the legacy-unprefixed read/delete fallback. + +Both scripts read the control DB then each tenant DB via raw SQL (no Flask app +context — the ORM's default bind is the wrong database for every tenant), and +the sync imports `collect_referenced` from the audit script so the two key sets +can never diverge. Neither writes to any database. + +`STORAGE_BACKEND` is process-wide, so cutover is all-tenants-at-once; isolation +comes from the prefix, not from separate backends. Per-tenant backend selection +would require the resolver to carry a storage selector and `get_backend()` to +cache per tenant rather than per app — do not half-build it. + +Rollback is one env var (`STORAGE_BACKEND=local` + restart); the sync never +deletes local files. + +**Open, deliberately not done in MT-20:** `routes/tenant_settings.py::_save_logo` +still writes tenant logos directly to `static/uploads/logos/` with `os.path.join`, +bypassing the seam. After cutover it is the only remaining local-disk writer, so +logos would sit outside whatever backs up R2. + --- ## 8. Tenant-zero (LT Services) migration diff --git a/requirements.txt b/requirements.txt index c15245d..9d6c28d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -25,3 +25,4 @@ groq stripe pyotp qrcode +boto3>=1.34 diff --git a/scripts/audit_photos.py b/scripts/audit_photos.py new file mode 100644 index 0000000..2e248c4 --- /dev/null +++ b/scripts/audit_photos.py @@ -0,0 +1,390 @@ +#!/usr/bin/env python3 +""" +scripts/audit_photos.py — Pre-cutover photo audit, PER TENANT (READ-ONLY) +=========================================================================== + +Multi-tenant port of the single-tenant script of the same name (ST §22 Phase 0). +Writes NOTHING to any database and NOTHING to the uploads tree. It only reads, +and emits a report. + +Why this differs from the single-tenant version +----------------------------------------------- +The local backend does NOT prefix keys (see app/utils/storage.py, "Tenant +isolation"), so every tenant's photos sit intermixed in one shared directory: + + app/static/uploads/issue_photos/.jpg + +A file on disk therefore carries no evidence of which tenant owns it. Ownership +is established the only way it can be — by walking each tenant's own database +and collecting the keys it references. That mapping is exactly what the R2 sync +needs, because on R2 the object lives at ``t/uploads/...``. + +What it does +------------ +1. For every tenant (control DB, status in provisioning/active/suspended), + collects every photo key REFERENCED by that tenant's DB, across all 5 sources: + - issues.photo_path (string) + - issues.mobile_photo_paths[] (JSON list — iPad evidence) + - issues.result_photos[] (JSON list — web resolution photos) + - inspections.form_data{} (any 'uploads/...' string, incl. nested) + - inspection_results.photo_path (string) + (Signatures are stored inline as 'data:' base64, NOT files — correctly ignored.) +2. Checks each referenced key against the file actually on disk. +3. Walks the uploads tree and finds files referenced by NO tenant (orphans). +4. Flags any key referenced by MORE THAN ONE tenant (should be impossible — + keys are uuid4 — but a restore/copy could produce one, and it would mean the + same bytes get written under two tenant prefixes). +5. Prints a summary and writes a detailed JSON report kept as the baseline. + +The number that matters, per tenant: **distinct referenced photos that resolve +on disk**. That count must be identical after cutover. Any "missing on disk" +here is a PRE-EXISTING broken reference — surfaced now so it can never be +blamed on the move. + +Usage +----- + cd /home/jqc/lt_janitorial_quality_control + source venv/bin/activate + python scripts/audit_photos.py # all tenants -> /tmp report + python scripts/audit_photos.py --tenant lts # one tenant (id or slug) + python scripts/audit_photos.py --report /home/jqc/photo_audit.json + python scripts/audit_photos.py --list-orphans # also print orphan paths + +Requires CONTROL_DATABASE_URL and CONTROL_FERNET_KEY in the environment (.env), +the same as any other control-plane command. Exit code is always 0 (audit is +informational). Nothing is modified. +""" + +import os +import sys +import json +import argparse +from collections import defaultdict +from datetime import datetime + +# Make the repo root importable when run from anywhere. +_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) +sys.path.insert(0, _REPO_ROOT) + +from dotenv import load_dotenv # noqa: E402 +load_dotenv(os.path.join(_REPO_ROOT, '.env')) + +from sqlalchemy import create_engine, text # noqa: E402 + +from control.base import control_session # noqa: E402 +from control.models import Tenant # noqa: E402 + +# A deleted tenant's DB may be gone; everything else is audited. +AUDITABLE_STATUSES = ('provisioning', 'active', 'suspended') + +STATIC_FOLDER = os.path.join(_REPO_ROOT, 'app', 'static') +UPLOADS_ROOT = os.path.join(STATIC_FOLDER, 'uploads') + + +# ── Path helpers ────────────────────────────────────────────────────────────── + +def normalize(path): + """ + Return a canonical 'uploads/...' key, or None if this value is not a stored + photo file (base64 signatures, unsynced iPad sentinels, remote URLs, blanks). + """ + if not isinstance(path, str): + return None + p = path.strip() + if not p: + return None + # Not on-disk files: + if p.startswith('data:') or p.startswith('local://'): + return None + if p.startswith('http://') or p.startswith('https://'): + return None + # Tolerate a few stored variants and reduce to 'uploads/...' + for prefix in ('/static/', 'static/', '/'): + if p.startswith(prefix): + p = p[len(prefix):] + if not p.startswith('uploads/'): + return None + return p + + +def walk_strings(value): + """Yield every string found in an arbitrarily nested JSON value (dict/list/str).""" + if isinstance(value, str): + yield value + elif isinstance(value, dict): + for v in value.values(): + yield from walk_strings(v) + elif isinstance(value, (list, tuple)): + for v in value: + yield from walk_strings(v) + + +def as_json(value): + """Decode a JSON column that may arrive as str (PyMySQL) or already-parsed.""" + if value is None: + return None + if isinstance(value, (dict, list)): + return value + if isinstance(value, (bytes, bytearray)): + value = value.decode('utf-8', 'replace') + if isinstance(value, str): + try: + return json.loads(value) + except (ValueError, TypeError): + return None + return None + + +# ── Reference collection (one tenant DB) ────────────────────────────────────── + +def collect_referenced(db_uri): + """ + Return (referenced, by_source) for a single tenant database. + + referenced : { normalized_key -> [ "issue:123:photo_path", ... ] } + by_source : { source_name -> count_of_refs } + + Raw SQL on purpose: this runs outside a Flask app context, so the ORM models + (bound to app.db) are not usable without dragging in create_app and its + default bind — which is the wrong database for every tenant. + """ + referenced = defaultdict(list) + by_source = defaultdict(int) + + def add(path, ref_label, source_name): + norm = normalize(path) + if norm is None: + return + referenced[norm].append(ref_label) + by_source[source_name] += 1 + + engine = create_engine(db_uri, pool_pre_ping=True, future=True) + try: + with engine.connect() as conn: + # --- Issues --- + rows = conn.execute(text( + 'SELECT id, photo_path, mobile_photo_paths, result_photos FROM issues' + )) + for rid, photo_path, mobile_paths, result_photos in rows: + add(photo_path, f'issue:{rid}:photo_path', 'issue.photo_path') + for pth in (as_json(mobile_paths) or []): + add(pth, f'issue:{rid}:mobile_photo_paths', 'issue.mobile_photo_paths') + for pth in (as_json(result_photos) or []): + add(pth, f'issue:{rid}:result_photos', 'issue.result_photos') + + # --- Inspections: form_data image fields (Inspection has NO photo_path) --- + rows = conn.execute(text('SELECT id, form_data FROM inspections')) + for rid, form_data in rows: + for s in walk_strings(as_json(form_data) or {}): + add(s, f'inspection:{rid}:form_data', 'inspection.form_data(image)') + + # --- InspectionResult: per checklist-item result photo --- + rows = conn.execute(text('SELECT id, photo_path FROM inspection_results')) + for rid, photo_path in rows: + add(photo_path, f'inspection_result:{rid}:photo_path', + 'inspection_result.photo_path') + finally: + engine.dispose() + + return referenced, by_source + + +# ── Disk scan ───────────────────────────────────────────────────────────────── + +def scan_disk(uploads_root): + """Return { relative_uploads_key -> size_bytes } for every file under uploads/.""" + on_disk = {} + if not os.path.isdir(uploads_root): + return on_disk + base = os.path.dirname(uploads_root) # .../static (so keys start with 'uploads/') + for dirpath, _dirs, files in os.walk(uploads_root): + for name in files: + full = os.path.join(dirpath, name) + rel = os.path.relpath(full, base).replace(os.sep, '/') + try: + on_disk[rel] = os.path.getsize(full) + except OSError: + on_disk[rel] = -1 + return on_disk + + +# ── Tenant selection ────────────────────────────────────────────────────────── + +def select_tenants(selector): + """Return [(id, slug, db_uri)] for 'all' or a single id/slug.""" + out = [] + with control_session() as s: + q = s.query(Tenant).filter(Tenant.status.in_(AUDITABLE_STATUSES)) + if selector and selector != 'all': + if str(selector).isdigit(): + q = q.filter(Tenant.id == int(selector)) + else: + q = q.filter(Tenant.slug == selector) + for t in q.order_by(Tenant.id).all(): + out.append((t.id, t.slug, t.db_uri)) # db_uri decrypts here + return out + + +# ── Main ────────────────────────────────────────────────────────────────────── + +def human(nbytes): + n = float(nbytes) + for unit in ('B', 'KB', 'MB', 'GB', 'TB'): + if abs(n) < 1024.0: + return f'{n:.1f} {unit}' + n /= 1024.0 + return f'{n:.1f} PB' + + +def main(): + parser = argparse.ArgumentParser( + description='Read-only per-tenant photo audit (R2 cutover, phase 0).') + parser.add_argument('--tenant', default='all', + help="Tenant id or slug, or 'all' (default).") + parser.add_argument('--report', default=None, + help='Path to write the JSON report (default: /tmp/jqc_photo_audit_.json)') + parser.add_argument('--list-orphans', action='store_true', + help='Also print orphan file paths to stdout') + args = parser.parse_args() + + tenants = select_tenants(args.tenant) + if not tenants: + print(f'No tenants matched --tenant {args.tenant!r} ' + f'(statuses: {", ".join(AUDITABLE_STATUSES)}).') + return + + on_disk = scan_disk(UPLOADS_ROOT) + on_disk_keys = set(on_disk.keys()) + + per_tenant = [] + claimed = defaultdict(list) # key -> [tenant slug, ...] (multi-claim detection) + + print('=' * 72) + print(' JQC PHOTO AUDIT — MULTI-TENANT (read-only — nothing was modified)') + print('=' * 72) + print(f' uploads root : {UPLOADS_ROOT}') + print(f' tenants : {len(tenants)}') + + for tid, slug, db_uri in tenants: + try: + referenced, by_source = collect_referenced(db_uri) + except Exception as exc: + print(f'\n-- tenant {tid} ({slug}) : ERROR reading DB: {exc}') + per_tenant.append({'tenant_id': tid, 'slug': slug, 'error': str(exc)}) + continue + + referenced_keys = set(referenced.keys()) + for k in referenced_keys: + claimed[k].append(slug) + + present = referenced_keys & on_disk_keys + missing = referenced_keys - on_disk_keys + present_bytes = sum(on_disk[k] for k in present if on_disk[k] > 0) + + per_tenant.append({ + 'tenant_id': tid, + 'slug': slug, + 'referenced': { + 'distinct_keys': len(referenced_keys), + 'by_source': dict(by_source), + 'present_on_disk': len(present), + 'present_bytes': present_bytes, + 'missing_on_disk': len(missing), + }, + 'missing_on_disk': sorted( + [{'key': k, 'referenced_by': sorted(set(referenced[k]))} for k in missing], + key=lambda d: d['key'] + ), + # The exact upload set for this tenant, consumed by + # scripts/migrate_photos_to_r2.py (which recomputes it itself). + 'r2_object_prefix': f't{tid}/', + }) + + print('-' * 72) + print(f' tenant {tid} ({slug})') + for src, cnt in sorted(by_source.items()): + print(f' {src:<34} {cnt:>8} refs') + print(f' {"distinct keys referenced":<34} {len(referenced_keys):>8}') + print(f' {"present on disk":<34} {len(present):>8} ' + f'({human(present_bytes)}) <-- must be unchanged after cutover') + print(f' {"MISSING on disk":<34} {len(missing):>8} ' + f'<-- pre-existing broken refs') + print(f' {"R2 object prefix":<34} {"t" + str(tid) + "/":>8}') + + all_referenced = set(claimed.keys()) + orphans = on_disk_keys - all_referenced + orphan_bytes = sum(on_disk[k] for k in orphans if on_disk[k] > 0) + total_bytes = sum(v for v in on_disk.values() if v > 0) + multi_claimed = {k: v for k, v in claimed.items() if len(v) > 1} + + # Per-subfolder disk breakdown + by_folder = defaultdict(lambda: [0, 0]) # folder -> [count, bytes] + for k, sz in on_disk.items(): + parts = k.split('/') + folder = parts[1] if len(parts) > 2 else '(root)' + by_folder[folder][0] += 1 + by_folder[folder][1] += sz if sz > 0 else 0 + + print('-' * 72) + print(' ON DISK (shared, unprefixed — local backend does not isolate)') + for folder, (cnt, b) in sorted(by_folder.items()): + print(f' {folder:<24} {cnt:>8} files {human(b):>12}') + print(f' {"TOTAL":<24} {len(on_disk):>8} files {human(total_bytes):>12}') + print('-' * 72) + print(' RECONCILIATION (all audited tenants)') + print(f' referenced by some tenant : {len(all_referenced):>8}') + print(f' on disk but UNREFERENCED : {len(orphans):>8} ' + f'({human(orphan_bytes)}, NOT synced — no tenant owns them)') + print(f' keys claimed by >1 tenant : {len(multi_claimed):>8} ' + f'<-- must be 0') + print('=' * 72) + + if multi_claimed: + print('\n WARNING: the following keys are referenced by more than one tenant.') + print(' They will be copied under EVERY claiming prefix. Investigate the') + print(' source (a DB restore across tenants?) before cutover:') + for k, slugs in list(multi_claimed.items())[:10]: + print(f' - {k} ({", ".join(slugs)})') + if len(multi_claimed) > 10: + print(f' ... and {len(multi_claimed) - 10} more (see JSON report)') + + if args.list_orphans and orphans: + print(f'\n Orphan files ({len(orphans)}):') + for k in sorted(orphans): + print(f' - {k} ({human(on_disk[k])})') + + report = { + 'generated_at': datetime.utcnow().isoformat() + 'Z', + 'static_folder': STATIC_FOLDER, + 'uploads_root': UPLOADS_ROOT, + 'tenant_selector': args.tenant, + 'tenants': per_tenant, + 'disk': { + 'total_files': len(on_disk), + 'total_bytes': total_bytes, + 'by_folder': {k: {'files': v[0], 'bytes': v[1]} for k, v in by_folder.items()}, + }, + 'orphans': { + 'count': len(orphans), + 'bytes': orphan_bytes, + 'files': sorted([{'key': k, 'bytes': on_disk[k]} for k in orphans], + key=lambda d: d['key']), + }, + 'multi_claimed': {k: sorted(v) for k, v in multi_claimed.items()}, + } + + report_path = args.report or os.path.join( + '/tmp', f'jqc_photo_audit_{datetime.utcnow().strftime("%Y%m%d_%H%M%S")}.json') + try: + with open(report_path, 'w') as fh: + json.dump(report, fh, indent=2) + print(f'\n Full JSON report written to: {report_path}') + except OSError as e: + print(f'\n Could not write report to {report_path}: {e}') + + baseline = sum(t.get('referenced', {}).get('present_on_disk', 0) for t in per_tenant) + print(f'\n BASELINE TO REMEMBER: {baseline} photos must still resolve after cutover.\n') + + +if __name__ == '__main__': + main() diff --git a/scripts/migrate_photos_to_r2.py b/scripts/migrate_photos_to_r2.py new file mode 100644 index 0000000..526ee05 --- /dev/null +++ b/scripts/migrate_photos_to_r2.py @@ -0,0 +1,329 @@ +#!/usr/bin/env python3 +""" +scripts/migrate_photos_to_r2.py — Verified per-tenant photo sync to R2 +======================================================================== + +Multi-tenant port of the single-tenant script of the same name (ST §22 Phase 3). +Copies each tenant's photo files from the shared local uploads tree to the R2 +bucket **under that tenant's object prefix**, and VERIFIES each one (MD5 + size). + +It is: + - COPY-ONLY : never deletes or modifies local files, never touches any DB. + - IDEMPOTENT : re-running skips objects already present and verified. + - RESUMABLE : safe to interrupt and re-run; picks up where it left off. + - GATED : exits non-zero if ANY file fails verification, so you never + flip STORAGE_BACKEND=s3 on an incomplete/corrupt copy. + +Key mapping — the whole point of this script +-------------------------------------------- +The DB stores an unprefixed key and always will: + + DB value : uploads/issue_photos/ab12….jpg + R2 object : t3/uploads/issue_photos/ab12….jpg + +The ``t/`` prefix is applied by ``S3Backend._object_key()`` at +runtime (app/utils/storage.py). This script must therefore write to the SAME +prefixed key, or nothing will resolve after cutover. + +Because the local backend does NOT prefix, one shared directory holds every +tenant's files and a file on disk carries no ownership marker. Ownership is +resolved exactly as in scripts/audit_photos.py: by reading each tenant's own +database and collecting the keys it references. Consequences, both intended: + + * Files referenced by NO tenant (orphans) are NOT uploaded — no tenant's + prefix could legitimately claim them, and nothing reads them. + * A key referenced by two tenants is copied under BOTH prefixes. Run + audit_photos.py first; it flags that case. + +R2 credentials come from the environment / .env: R2_ENDPOINT_URL, +R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY, R2_BUCKET. Control-plane access needs +CONTROL_DATABASE_URL and CONTROL_FERNET_KEY. Run this while the app is still +serving on STORAGE_BACKEND=local — it only writes to R2. + +Why put_object (not upload_file): a single-part PUT makes the R2 ETag equal the +object's MD5, so verification is a direct hash comparison. Photos are <= 50 MB +(MAX_CONTENT_LENGTH), well within a single PUT. + +Usage +----- + cd /home/jqc/lt_janitorial_quality_control + source venv/bin/activate + pip install boto3 # if not already + + python scripts/migrate_photos_to_r2.py --dry-run # plan only + python scripts/migrate_photos_to_r2.py # all tenants + python scripts/migrate_photos_to_r2.py --tenant lts # one tenant + python scripts/migrate_photos_to_r2.py # again at cutover (delta) + python scripts/migrate_photos_to_r2.py --force # re-upload everything + python scripts/migrate_photos_to_r2.py --log /home/jqc/r2_sync.json + +Exit code: 0 only when every referenced file that exists on disk is verified in +R2 under its tenant prefix, with zero mismatches and zero errors. Non-zero +otherwise. References missing on disk are reported and do NOT fail the gate — +they were already broken before this move (audit_photos.py is the baseline). +""" + +import os +import sys +import json +import hashlib +import argparse +from datetime import datetime + +_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) +sys.path.insert(0, _REPO_ROOT) + +from dotenv import load_dotenv # noqa: E402 +load_dotenv(os.path.join(_REPO_ROOT, '.env')) + +# Reuse the audit script's reference collection so the two can never diverge. +from scripts.audit_photos import ( # noqa: E402 + AUDITABLE_STATUSES, + STATIC_FOLDER, + UPLOADS_ROOT, + collect_referenced, + select_tenants, +) + +_CONTENT_TYPES = { + 'jpg': 'image/jpeg', 'jpeg': 'image/jpeg', + 'png': 'image/png', 'gif': 'image/gif', +} +_CHUNK = 1024 * 1024 # 1 MB + + +def _md5_and_size(path): + h = hashlib.md5() + size = 0 + with open(path, 'rb') as fh: + while True: + chunk = fh.read(_CHUNK) + if not chunk: + break + h.update(chunk) + size += len(chunk) + return h.hexdigest(), size + + +def _content_type(key): + ext = key.rsplit('.', 1)[-1].lower() if '.' in key else '' + return _CONTENT_TYPES.get(ext, 'application/octet-stream') + + +def _remote_state(client, bucket, key): + """Return (exists, etag_no_quotes, size) for a key in the bucket.""" + try: + resp = client.head_object(Bucket=bucket, Key=key) + return True, resp['ETag'].strip('"'), resp['ContentLength'] + except Exception: + return False, None, None + + +def human(n): + n = float(n) + for unit in ('B', 'KB', 'MB', 'GB', 'TB'): + if abs(n) < 1024.0: + return f'{n:.1f} {unit}' + n /= 1024.0 + return f'{n:.1f} PB' + + +def main(): + parser = argparse.ArgumentParser( + description='Verified per-tenant photo sync to R2 (prefix t/).') + parser.add_argument('--tenant', default='all', + help="Tenant id or slug, or 'all' (default).") + parser.add_argument('--dry-run', action='store_true', + help='Report what would happen; upload nothing.') + parser.add_argument('--force', action='store_true', + help='Re-upload even if the object already verifies.') + parser.add_argument('--log', default=None, + help='Path to write the JSON result log (default: /tmp/...).') + args = parser.parse_args() + + missing_cfg = [k for k in ('R2_ENDPOINT_URL', 'R2_ACCESS_KEY_ID', + 'R2_SECRET_ACCESS_KEY', 'R2_BUCKET') + if not os.environ.get(k)] + if missing_cfg: + print(f'ERROR: missing R2 config: {", ".join(missing_cfg)}') + print('Set them in .env before running the sync.') + sys.exit(2) + + try: + import boto3 + from botocore.config import Config as BotoConfig + except ImportError: + print('ERROR: boto3 not installed. Run: pip install boto3') + sys.exit(2) + + bucket = os.environ['R2_BUCKET'] + client = boto3.client( + 's3', + endpoint_url = os.environ['R2_ENDPOINT_URL'], + aws_access_key_id = os.environ['R2_ACCESS_KEY_ID'], + aws_secret_access_key = os.environ['R2_SECRET_ACCESS_KEY'], + region_name = 'auto', + config = BotoConfig(signature_version='s3v4'), + ) + + if not os.path.isdir(UPLOADS_ROOT): + print(f'ERROR: uploads folder not found: {UPLOADS_ROOT}') + sys.exit(2) + + tenants = select_tenants(args.tenant) + if not tenants: + print(f'No tenants matched --tenant {args.tenant!r} ' + f'(statuses: {", ".join(AUDITABLE_STATUSES)}).') + sys.exit(2) + + totals = {'referenced': 0, 'missing_on_disk': 0, 'skipped_verified': 0, + 'uploaded': 0, 'verify_failed': 0, 'errors': 0, 'bytes_uploaded': 0} + failures = [] + per_tenant = [] + + print('=' * 72) + print(f' R2 PHOTO SYNC — MULTI-TENANT{" (DRY RUN)" if args.dry_run else ""}') + print(f' bucket : {bucket}') + print(f' source : {UPLOADS_ROOT}') + print(f' tenants : {len(tenants)}') + print('=' * 72) + + for tid, slug, db_uri in tenants: + prefix = f't{tid}/' + counts = {'referenced': 0, 'missing_on_disk': 0, 'skipped_verified': 0, + 'uploaded': 0, 'verify_failed': 0, 'errors': 0, 'bytes_uploaded': 0} + + print(f'-- tenant {tid} ({slug}) prefix={prefix}') + + try: + referenced, _by_source = collect_referenced(db_uri) + except Exception as exc: + print(f' ERROR reading tenant DB: {exc}') + totals['errors'] += 1 + failures.append({'tenant_id': tid, 'slug': slug, + 'stage': 'tenant_db', 'error': str(exc)}) + per_tenant.append({'tenant_id': tid, 'slug': slug, 'error': str(exc)}) + continue + + for key in sorted(referenced.keys()): + counts['referenced'] += 1 + abs_path = os.path.normpath(os.path.join(STATIC_FOLDER, key)) + + if not os.path.isfile(abs_path): + # Pre-existing broken reference — surfaced by audit_photos.py. + counts['missing_on_disk'] += 1 + continue + + try: + local_md5, local_size = _md5_and_size(abs_path) + except OSError as e: + counts['errors'] += 1 + failures.append({'tenant_id': tid, 'key': key, + 'stage': 'read', 'error': str(e)}) + print(f' ERROR read {key}: {e}') + continue + + object_key = f'{prefix}{key}' + + # Resumable skip: already present and matching? + if not args.force: + exists, r_etag, r_size = _remote_state(client, bucket, object_key) + if exists and r_etag == local_md5 and r_size == local_size: + counts['skipped_verified'] += 1 + continue + + if args.dry_run: + counts['uploaded'] += 1 # would upload + print(f' WOULD PUT {object_key} ({human(local_size)})') + continue + + # Upload (single-part PUT so ETag == MD5) + try: + with open(abs_path, 'rb') as body: + client.put_object( + Bucket=bucket, Key=object_key, Body=body, + ContentType=_content_type(key), + ) + except Exception as e: + counts['errors'] += 1 + failures.append({'tenant_id': tid, 'key': object_key, + 'stage': 'upload', 'error': str(e)}) + print(f' ERROR upload {object_key}: {e}') + continue + + # Verify: re-HEAD and compare ETag(MD5) + size + exists, r_etag, r_size = _remote_state(client, bucket, object_key) + if exists and r_etag == local_md5 and r_size == local_size: + counts['uploaded'] += 1 + counts['bytes_uploaded'] += local_size + else: + counts['verify_failed'] += 1 + failures.append({ + 'tenant_id': tid, 'key': object_key, 'stage': 'verify', + 'local_md5': local_md5, 'local_size': local_size, + 'remote_etag': r_etag, 'remote_size': r_size, + }) + print(f' VERIFY FAIL {object_key} local_md5={local_md5} ' + f'remote_etag={r_etag} local_size={local_size} remote_size={r_size}') + + for k in totals: + totals[k] += counts[k] + per_tenant.append({'tenant_id': tid, 'slug': slug, + 'prefix': prefix, 'counts': counts}) + + verified_here = counts['skipped_verified'] + (0 if args.dry_run else counts['uploaded']) + print(f' referenced={counts["referenced"]} ' + f'missing_on_disk={counts["missing_on_disk"]} ' + f'verified={verified_here} ' + f'uploaded={counts["uploaded"]} ({human(counts["bytes_uploaded"])}) ' + f'verify_failed={counts["verify_failed"]} errors={counts["errors"]}') + + # ── summary ── + print('-' * 72) + if args.dry_run: + action_line = f' would upload : {totals["uploaded"]}' + else: + action_line = (f' uploaded + verified : {totals["uploaded"]}' + f' ({human(totals["bytes_uploaded"])})') + print(f' referenced keys (all tenants) : {totals["referenced"]}') + print(f' referenced but missing on disk: {totals["missing_on_disk"]} ' + f'(pre-existing broken refs)') + print(f' already verified (skipped) : {totals["skipped_verified"]}') + print(action_line) + print(f' verify failures : {totals["verify_failed"]}') + print(f' errors : {totals["errors"]}') + print('=' * 72) + + verified_total = totals['skipped_verified'] + (0 if args.dry_run else totals['uploaded']) + syncable = totals['referenced'] - totals['missing_on_disk'] + safe = (totals['verify_failed'] == 0 and totals['errors'] == 0) + + if args.dry_run: + print(' DRY RUN — nothing uploaded. Re-run without --dry-run to sync.') + elif safe: + print(f' GATE: {verified_total}/{syncable} referenced files verified in R2 under ' + f'their tenant prefix, 0 mismatches, 0 errors.') + print(' ✅ SAFE to flip STORAGE_BACKEND=s3. Local files are untouched.') + else: + print(f' GATE: {totals["verify_failed"]} verify failure(s), {totals["errors"]} error(s).') + print(' ❌ NOT safe to cut over. Investigate failures (see log), then re-run.') + + log_path = args.log or os.path.join( + '/tmp', f'jqc_r2_sync_{datetime.utcnow().strftime("%Y%m%d_%H%M%S")}.json') + try: + with open(log_path, 'w') as fh: + json.dump({ + 'generated_at': datetime.utcnow().isoformat() + 'Z', + 'bucket': bucket, 'dry_run': args.dry_run, + 'tenant_selector': args.tenant, + 'totals': totals, 'tenants': per_tenant, 'failures': failures, + }, fh, indent=2) + print(f'\n Log written to: {log_path}') + except OSError as e: + print(f'\n Could not write log to {log_path}: {e}') + + sys.exit(0 if safe else 1) + + +if __name__ == '__main__': + main() diff --git a/tests/test_storage_backend.py b/tests/test_storage_backend.py new file mode 100644 index 0000000..668890a --- /dev/null +++ b/tests/test_storage_backend.py @@ -0,0 +1,154 @@ +""" +tests/test_storage_backend.py +----------------------------- +Guards the one thing that makes S3 storage safe in a multi-tenant deployment: +the ``t/`` object-key prefix (app/utils/storage.py). + +Two failure modes are pinned here, both of which would be silent in production: + + 1. A prefix regression makes tenant A's presigned URL point at tenant B's + object — a cross-tenant data leak with no error anywhere. + 2. An upload reaching S3Backend.save() with no tenant bound writes an + unprefixed key. Nothing would break at write time; the leak appears later + when a second tenant's DB happens to hold the same key string. + +S3Backend is constructed with ``object.__new__`` and hand-wired attributes on +purpose: its ``__init__`` builds a real boto3 client, and these tests are about +key arithmetic, not the network. A stub client records the calls. +""" + +import pytest +from flask import g + +from app.utils import storage + + +class _StubClient: + """Minimal S3 client double — records calls, answers HEAD from a key set.""" + + def __init__(self, existing=()): + self.existing = set(existing) + self.put_calls = [] + self.presigned = [] + self.deleted = [] + + def put_object(self, Bucket=None, Key=None, Body=None, ContentType=None): + self.put_calls.append(Key) + self.existing.add(Key) + return {} + + def head_object(self, Bucket=None, Key=None): + if Key in self.existing: + return {'ETag': '"x"', 'ContentLength': 1} + raise RuntimeError('404') + + def get_object(self, Bucket=None, Key=None): + raise RuntimeError('404') + + def generate_presigned_url(self, _op, Params=None, ExpiresIn=None): + key = (Params or {}).get('Key') + self.presigned.append(key) + return f'https://r2.example/{key}' + + def delete_object(self, Bucket=None, Key=None): + self.deleted.append(Key) + return {} + + +class _FakeTenant: + def __init__(self, tid): + self.id = tid + + +class _FakeUpload: + """Stands in for a werkzeug FileStorage well enough for save().""" + + def __init__(self, filename='photo.jpg', data=b'\xff\xd8\xffbytes'): + import io + self.filename = filename + self.stream = io.BytesIO(data) + + +def _make_s3_backend(stub): + backend = object.__new__(storage.S3Backend) + backend.bucket = 'test-bucket' + backend.ttl = 60 + backend.fallback = False + backend._client = stub + backend._local = storage.LocalBackend() + return backend + + +@pytest.fixture +def mt_ctx(app): + """App context with multi-tenancy on and tenant 7 bound.""" + with app.test_request_context('/'): + app.config['MULTI_TENANT_ENABLED'] = True + g.tenant = _FakeTenant(7) + try: + yield app + finally: + app.config['MULTI_TENANT_ENABLED'] = False + + +def test_tenant_key_prefix_reflects_bound_tenant(mt_ctx): + assert storage.tenant_key_prefix() == 't7/' + + +def test_tenant_key_prefix_empty_when_mt_disabled(app): + with app.test_request_context('/'): + app.config['MULTI_TENANT_ENABLED'] = False + g.tenant = _FakeTenant(7) + assert storage.tenant_key_prefix() == '' + + +def test_save_writes_prefixed_object_but_returns_bare_key(mt_ctx): + stub = backend_stub = _StubClient() + backend = _make_s3_backend(backend_stub) + + key = backend.save(_FakeUpload('evidence.JPG'), 'issue_photos') + + # DB value stays unprefixed and portable. + assert key.startswith('uploads/issue_photos/') + assert key.endswith('.jpg') + assert not key.startswith('t7/') + # The object on the wire is tenant-scoped. + assert stub.put_calls == [f't7/{key}'] + + +def test_save_refuses_when_no_tenant_bound(app): + """An upload path outside tenant resolution must fail loudly, not silently + write a key another tenant could later collide with.""" + with app.test_request_context('/'): + app.config['MULTI_TENANT_ENABLED'] = True + g.tenant = None + stub = _StubClient() + backend = _make_s3_backend(stub) + try: + with pytest.raises(RuntimeError): + backend.save(_FakeUpload(), 'issue_photos') + finally: + app.config['MULTI_TENANT_ENABLED'] = False + assert stub.put_calls == [] + + +def test_media_url_presigns_the_prefixed_key(mt_ctx): + stub = _StubClient() + backend = _make_s3_backend(stub) + + url = backend.media_url('uploads/issue_photos/abc.jpg') + + assert stub.presigned == ['t7/uploads/issue_photos/abc.jpg'] + assert url.endswith('t7/uploads/issue_photos/abc.jpg') + + +def test_exists_and_delete_cover_legacy_unprefixed_key(mt_ctx): + """Objects written before the prefix existed must stay reachable.""" + key = 'uploads/issue_photos/legacy.jpg' + stub = _StubClient(existing={key}) # only the unprefixed object + backend = _make_s3_backend(stub) + + assert backend.exists(key) is True + + backend.delete(key) + assert stub.deleted == [f't7/{key}', key]