Storage

Design

OpenTremor uses a content-addressed, multi-tenant storage model. Every collection is scoped to an org_id — either directly, or via a composite _id that starts with it ("{org_id}:...").

organizations        { _id: org_id, name, plan_tier, quota }
users                 { _id: user_id, email, password_hash, auth_provider }
memberships           { _id: "org:user", role }
resources             { _id: "org:hash", org_id, hash, value, analysis }
namespace_entries     { _id: "org:ns:hash", org_id, namespace, hash }
api_keys              { _id: key_id, org_id, key_hash, role, is_active }
installed_analyzers   { _id: "org:analyzer", org_id, analyzer_name, enabled }
custom_rules          { _id: "org:rule_id", org_id, rule_id, title, description, category, severity, analyzers[], enabled, requires_review, created_by }
rule_categories        { _id: "org:category_id", org_id, category_id, name, is_builtin_seed, created_by }
teams                  { _id: "org:team_id", org_id, team_id, name, created_by }
team_members           { _id: "org:team_id:user_id", org_id, team_id, user_id, added_by, added_at }
report_templates      { _id: "org:template_id", org_id, template_id, name, format, source, is_default, is_builtin_seed, created_by }
llm_credentials       { _id: "org:provider", org_id, provider, encrypted_key }
jobs                  { _id: job_id, org_id, namespace, analyzer, status, mode, trigger }
usage_events          { org_id, type, quantity, cost_usd, timestamp (TTL) }
findings              { _id: "org:hash:rule_id", org_id, resource_hash, rule_id, status, occurrence_count }
integrations          { _id: "org:provider", org_id, provider, installation_id, config, github_app }  # provider="github"
                      { _id: "org:provider", org_id, provider, url, encrypted_secret, enabled }        # provider="review_webhook"
github_manifest_states { _id: state_id, state_hash, org_id, user_id, expires_at (TTL, per-doc) }
report_links          { _id: link_id, token_hash, org_id, namespace, expires_at (TTL, per-doc) }
org_invites           { _id: invite_id, token_hash, org_id, role, revoked, expires_at (TTL, per-doc) }
spans                 { span_id, trace_id, org_id, type, timestamp (TTL), key_id, user_id }
audit_events          { event_id, org_id (nullable), action, target_type, target_id, actor_user_id, actor_key_id, actor_email, before, after, timestamp (TTL) }
rate_limits           { _id: "policy:identity:window", bucket, count, expires_at (TTL, per-doc) }

See Architecture — Data model for the full entity-relationship diagram and the reasoning behind each collection.

Why two collections for resources?

Concern	Decision
Same analysis unit in N namespaces	Store value once; create N lightweight entries
Analysis reuse	analysis lives on the resource — same hash = same analysis everywhere
Namespace queries	namespace_entries is a small, indexed collection; no full scan needed
Content deduplication	SHA-256 hash of normalised content, per org — resources._id = "{org_id}:{hash}"

Dedup deliberately stops at the org boundary: the same byte-identical input submitted by two different orgs is stored and analysed independently, since value.body can itself carry tenant-identifying data.

---

Backends

Backend	Class	Persistence	Recommended for
MongoDB	MongoBackend	Yes	Production
In-memory	InMemoryStorage	No	Tests, local dev

Both implement the same StorageBackend abstract class and are interchangeable at the dependency injection level.

---

StorageBackend interface

class StorageBackend(ABC):
    # Organizations / users / memberships
    async def create_organization(self, doc: dict) -> None: ...
    async def get_organization(self, org_id: str) -> dict | None: ...
    async def update_organization(self, org_id: str, fields: dict) -> None: ...
    async def create_user(self, doc: dict) -> None: ...
    async def create_membership(self, doc: dict) -> None: ...

    # Resources (every method org_id-scoped)
    async def create_resource(self, org_id: str, hash: str, value: dict) -> None: ...
    async def read_resource(self, org_id: str, hash: str) -> dict | None: ...
    async def set_analysis(self, org_id: str, hash: str, analysis: dict) -> None: ...
    async def add_to_namespace(self, org_id: str, hash: str, namespace: str) -> None: ...
    async def find_by_namespace(self, org_id: str, namespace: str) -> list[dict]: ...
    async def find_next_unanalysed(self, org_id: str, namespace: str) -> dict | None: ...

    # Auth
    async def create_api_key(self, doc: dict) -> None: ...
    async def get_api_key_by_hash(self, key_hash: str) -> dict | None: ...
    async def list_api_keys(self, org_id: str) -> list[dict]: ...
    async def delete_api_key(self, org_id: str, key_id: str) -> bool: ...

    # Analyzer install / custom rules / LLM credentials / jobs / findings / integrations
    async def list_installed_analyzers(self, org_id: str) -> list[dict]: ...
    async def create_custom_rule(self, doc: dict) -> None: ...
    async def get_custom_rule(self, org_id: str, rule_id: str) -> dict | None: ...
    async def list_custom_rules(self, org_id: str, analyzer: str | None = None, category: str | None = None, enabled: bool | None = None) -> list[dict]: ...
    async def update_custom_rule(self, org_id: str, rule_id: str, fields: dict) -> None: ...
    async def delete_custom_rule(self, org_id: str, rule_id: str) -> bool: ...
    async def create_rule_category(self, doc: dict) -> None: ...
    async def get_rule_category(self, org_id: str, category_id: str) -> dict | None: ...
    async def list_rule_categories(self, org_id: str) -> list[dict]: ...
    async def update_rule_category(self, org_id: str, category_id: str, fields: dict) -> dict | None: ...
    async def delete_rule_category(self, org_id: str, category_id: str) -> bool: ...
    async def create_team(self, doc: dict) -> None: ...
    async def get_team(self, org_id: str, team_id: str) -> dict | None: ...
    async def list_teams(self, org_id: str) -> list[dict]: ...
    async def update_team(self, org_id: str, team_id: str, fields: dict) -> dict | None: ...
    async def delete_team(self, org_id: str, team_id: str) -> bool: ...
    async def add_team_member(self, doc: dict) -> None: ...
    async def remove_team_member(self, org_id: str, team_id: str, user_id: str) -> bool: ...
    async def list_team_members(self, org_id: str, team_id: str) -> list[dict]: ...
    async def list_all_team_memberships(self, org_id: str) -> list[dict]: ...
    async def delete_team_members_for_team(self, org_id: str, team_id: str) -> None: ...
    async def create_report_template(self, doc: dict) -> None: ...
    async def get_report_template(self, org_id: str, template_id: str) -> dict | None: ...
    async def list_report_templates(self, org_id: str, format: str | None = None, is_default: bool | None = None) -> list[dict]: ...
    async def update_report_template(self, org_id: str, template_id: str, fields: dict) -> dict | None: ...
    async def delete_report_template(self, org_id: str, template_id: str) -> bool: ...
    async def create_llm_credential(self, doc: dict) -> None: ...
    async def create_job(self, doc: dict) -> None: ...
    async def upsert_finding(self, org_id: str, resource_hash: str, finding: dict) -> None: ...
    async def create_integration(self, doc: dict) -> None: ...

    # Report links (public HTML report capability tokens)
    async def create_report_link(self, doc: dict) -> None: ...
    async def get_report_link_by_token_hash(self, token_hash: str) -> dict | None: ...

    # Org invitation links (reusable-until-expiry join capability tokens)
    async def create_invite(self, doc: dict) -> None: ...
    async def get_invite_by_token_hash(self, token_hash: str) -> dict | None: ...
    async def list_invites(self, org_id: str) -> list[dict]: ...
    async def revoke_invite(self, org_id: str, invite_id: str) -> bool: ...

    # Billing
    async def record_usage_event(self, doc: dict) -> None: ...
    async def get_monthly_llm_cost(self, org_id: str) -> float: ...

    # Telemetry (org_id=None means "all orgs")
    async def record_span(self, span: dict) -> None: ...
    async def query_spans(self, org_id, type_filter, user_id, key_id, limit, offset) -> list[dict]: ...
    async def get_metrics_summary(self, org_id, days: int) -> dict: ...
    async def get_key_usage(self, org_id, days: int) -> list[dict]: ...

    # Audit log (org_id=None means "all orgs")
    async def record_audit_event(self, event: dict) -> None: ...
    async def query_audit_events(self, org_id, action, actor_id, target_type, since, until, limit, offset) -> list[dict]: ...

    # Rate limiting — a shared counter, so budgets are per deployment not per replica.
    # The window arithmetic lives in libs/rate_limit.py; a backend only counts.
    async def hit_rate_limit(self, key: str, *, bucket: str, expires_at) -> int: ...
    async def peek_rate_limit(self, key: str) -> int: ...
    async def clear_rate_limit(self, bucket: str) -> None: ...

    async def setup(self) -> None: ...
    async def close(self) -> None: ...

All methods are async. setup() is called once at application startup (creates indexes). This is a trimmed view — see controllers/storage/base.py for the complete signature list, including pagination/filter parameters on list_findings/query_spans/query_audit_events.

---

Dependency injection

Route handlers never import a concrete backend class, and always thread the caller’s org_id through:

async def get_next_resource(
    namespace: str,
    auth: AuthContext = Depends(require_member),
    storage: StorageBackend = Depends(get_storage),
):
    resource = await storage.find_next_unanalysed(auth.org_id, namespace)