Storage Layer

The ABC is split by domain: PlatformStorageBackend (identity, orgs, admin, observability) and ProductStorageBackend (resources, findings, jobs, analysis configuration). StorageBackend is simply both, and one MongoBackend / one InMemoryStorage implement it — one connection and one database still serve both halves. The split exists so the multi-tenant half can be depended on without the analysis half; to back only that, subclass PlatformStorageBackend alone.

classDiagram
    class StorageLifecycle {
        <>
        +setup()
        +close()
    }
    class PlatformStorageBackend {
        <>
        +create_organization() / get_organization() / update_organization()
        +create_user() / get_user() / get_user_by_email()
        +create_membership() / get_membership() / list_members()
        +create_api_key() / get_api_key_by_hash()
        +list_api_keys(org_id) / delete_api_key(org_id, key_id)
        +create_team() / list_teams() / add_team_member()
        +create_invite() / get_invite_by_token_hash() / revoke_invite()
        +create_sso_connection() / create_scim_config()
        +get_platform_settings() / set_platform_settings()
        +record_usage_event() / list_usage_events() / get_monthly_llm_cost(org_id)
        +record_span() / query_spans() / get_metrics_summary() / get_key_usage()
        +record_audit_event() / query_audit_events() "org_id=None means all orgs"
        +apply_retention_ttls(retention) "collMod on metrics/usage_events/audit_events ttl_idx -- no-op on InMemoryStorage"
        +get_retention_sweep_state() / set_retention_sweep_state() "the sweep primitive the product registers steps into"
        +create_job() / get_job() / update_job() / list_jobs() / count_jobs() "update_job always stamps updated_at = now, in addition to whatever fields it's given"
        +sweep_stuck_jobs(older_than) "marks jobs with no progress update since older_than as failed, across every org"
        +get_job_sweep_state() / set_job_sweep_state()
        +delete_organization(org_id) "call last, after every bulk delete"
        +delete_X_for_org(org_id) "platform half: memberships/api_keys/team_members/teams/org_invites/jobs"
    }
    class ProductStorageBackend {
        <>
        +create_resource(org_id, hash, value)
        +read_resource(org_id, hash)
        +set_analysis(org_id, hash, analysis)
        +list_installed_analyzers(org_id) / set_analyzer_enabled()
        +create_custom_rule() / get_custom_rule() / list_custom_rules() / update_custom_rule() / delete_custom_rule()
        +create_report_template() / get_report_template() / list_report_templates() / update_report_template() / delete_report_template()
        +create_llm_credential() / get_llm_credential() / list_llm_credentials() / delete_llm_credential()
        +upsert_finding(initial_status) returns created:bool / get_finding() / list_findings() / list_findings_by_hashes() / update_finding_status() / get_findings_summary()
        +create_integration() / get_integration() / get_integration_by_installation_id() / delete_integration()
        +purge_findings(statuses, severities, older_than) "callers only ever pass terminal statuses"
        +redact_stale_resources(older_than) "overwrites value, leaves analysis/hash intact"
        +delete_X_for_org(org_id) "product half: resources/namespace_entries/installed_analyzers/custom_rules/report_templates/rule_categories/llm_credentials/findings"
    }
    class StorageBackend {
        <>
        +inherits both halves "what a single-process OpenTremor deployment needs"
    }
    class InMemoryStorage {
        -dict _orgs
        -dict _users
        -dict _memberships
        -dict _resources "keyed by org_id:hash, incl. last_ingested_at"
        -dict _entries "keyed by (org_id, namespace)"
        -dict _api_keys
        -dict _llm_credentials
        -dict _jobs
        -list _usage_events
        -dict _findings "keyed by (org_id, hash, rule_id)"
        -dict _integrations "keyed by (org_id, provider)"
        -dict _manifest_states "keyed by state_id, looked up by state_hash"
        -deque _spans
        -deque _audit_events "maxlen 50k, bigger than _spans' 10k -- lower volume per request"
        -dict _retention_sweep_state "None until a sweep has ever run"
        -dict _jobs "job_id -> doc (platform half)"
        -dict _job_sweep_state "None until a stuck-job sweep has ever run (platform half)"
    }
    class MongoBackend {
        -Collection orgs_col
        -Collection users_col
        -Collection memberships_col
        -Collection res_col "_id = org_id:hash"
        -Collection ent_col "_id = org_id:namespace:hash"
        -Collection keys_col
        -Collection llm_credentials_col "_id = org_id:provider"
        -Collection usage_events_col "TTL-indexed"
        -Collection findings_col "_id = org_id:hash:rule_id"
        -Collection integrations_col "_id = org_id:provider"
        -Collection github_manifest_states_col "unique idx on state_hash, TTL on expires_at"
        -Collection metrics_col
        -Collection audit_events_col "TTL-indexed on its own audit_retention_days -- no longer shares retention_days with metrics_col"
        -Collection retention_sweep_state_col "singleton, fixed _id"
        -Collection jobs_col "platform half"
        -Collection job_sweep_state_col "singleton, fixed _id (platform half)"
        -int retention_days "telemetry/usage TTL, seconds"
        -int audit_retention_days
    }
    StorageLifecycle <|-- PlatformStorageBackend
    StorageLifecycle <|-- ProductStorageBackend
    PlatformStorageBackend <|-- StorageBackend
    ProductStorageBackend <|-- StorageBackend
    StorageBackend <|-- InMemoryStorage : implements
    StorageBackend <|-- MongoBackend : implements

usage_events and audit_events are deliberately excluded from the org-offboarding cascade — orphaned by org_id rather than deleted, so the audit trail survives the thing it describes.

Route handlers only ever see StorageBackend through FastAPI’s Depends — nothing in a router imports InMemoryStorage or MongoBackend directly, which is what makes swapping the live backend (see POST /admin/storage/switch above) possible without touching route code.

create_integration/get_integration/delete_integration are reused verbatim for the needs-review webhook (provider="review_webhook") — no new collection or storage method was needed, since it’s the same (org_id, provider) shape GitHub already uses. That reuse is what let outbound webhooks become a platform primitive (send_org_webhook) without a schema of their own: provider doubles as the webhook’s name, so one org can hold several independent endpoints.

---