mongo-init.js

// Benthos Mongo bootstrap — runs ONCE on first container init.
// Schema changes after first init require a migration plan (Stage 2+).
// Defense-in-depth: $jsonSchema validators mirror the Pydantic models in
// benthos_api/models/. Drift is a known risk; lockstep is enforced by review.

const dbName = process.env.MONGO_INITDB_DATABASE || "benthos";
const appPassword = process.env.MONGO_APP_PASSWORD;
const vaultPassword = process.env.MONGO_VAULT_PASSWORD;

if (!appPassword || !vaultPassword) {
  throw new Error("MONGO_APP_PASSWORD and MONGO_VAULT_PASSWORD must be set");
}

const target = db.getSiblingDB(dbName);

const numericTypes = ["double", "int", "long", "decimal"];
const optionalNumeric = numericTypes.concat(["null"]);

// ---------------------------------------------------------------------------
// catch_events — Stage 1 fishing trip records (camada bot, L2)
// ---------------------------------------------------------------------------
target.createCollection("catch_events", {
  validator: {
    $jsonSchema: {
      bsonType: "object",
      required: [
        "schema_version", "catch_event_id", "community", "fisher_pseudo_id",
        "event_date", "consent_record_id",
        "catch_species", "catch_weight_kg", "catch_count",
        "fishing_effort_hours", "cpue_kg_per_hour", "created_at"
      ],
      additionalProperties: false,
      properties: {
        _id: {},
        schema_version:        { enum: ["1.0.0"] },
        catch_event_id:        { bsonType: "string", pattern: "^cev_[a-f0-9]{16}$" },
        community:             { bsonType: "string", pattern: "^[a-z]{2}-[a-z0-9-]+$" },
        fisher_pseudo_id:      { bsonType: "string", pattern: "^psd_[a-f0-9]{16}$" },
        event_date:            { bsonType: "date" },
        consent_record_id:     { bsonType: "string", pattern: "^csr_[A-Za-z0-9_-]+$" },
        catch_species:         { bsonType: "string", minLength: 1, maxLength: 200 },
        vernacular_name:       { bsonType: ["string", "null"], maxLength: 200 },
        catch_weight_kg:       { bsonType: numericTypes, minimum: 0, maximum: 100000 },
        catch_count:           { bsonType: ["int", "long"], minimum: 0 },
        catch_price_per_kg:    { bsonType: optionalNumeric, minimum: 0 },
        catch_price_currency:  { bsonType: ["string", "null"], pattern: "^[A-Z]{3}$" },
        fishing_effort_hours:  { bsonType: numericTypes, minimum: 0, exclusiveMinimum: true, maximum: 720 },
        cpue_kg_per_hour:      { bsonType: numericTypes, minimum: 0 },
        created_at:            { bsonType: "date" }
      }
    }
  },
  validationLevel: "strict",
  validationAction: "error"
});

// ---------------------------------------------------------------------------
// consent_records — vault collection (CARE-aligned)
// ---------------------------------------------------------------------------
target.createCollection("consent_records", {
  validator: {
    $jsonSchema: {
      bsonType: "object",
      required: [
        "schema_version", "consent_record_id", "community",
        "self_identification_frame", "consent_scope", "approved_categories",
        "data_custodian_entity", "prior_informed_consent_date",
        "approved_by", "data_retention_period_months",
        "withdrawal_mechanism", "revoked", "created_at"
      ],
      additionalProperties: false,
      properties: {
        _id: {},
        schema_version:    { enum: ["1.0.0"] },
        consent_record_id: { bsonType: "string", pattern: "^csr_[A-Za-z0-9_-]+$" },
        community:         { bsonType: "string", pattern: "^[a-z]{2}-[a-z0-9-]+$" },
        self_identification_frame: {
          enum: [
            "indigenous",
            "traditional_knowledge_custodian",
            "territorial_steward",
            "artisanal_producer_collective"
          ]
        },
        consent_scope: {
          enum: ["internal_only", "aggregate_research", "aggregate_funder", "named_public"]
        },
        approved_categories: {
          bsonType: "array",
          minItems: 1,
          uniqueItems: true,
          items: {
            enum: ["identity", "location", "catch", "observation",
                   "governance", "mett_score", "carbon_finance",
                   "traditional_knowledge", "community_voice"]
          }
        },
        data_custodian_entity:        { bsonType: "string", minLength: 1, maxLength: 500 },
        prior_informed_consent_date:  { bsonType: "date" },
        approved_by: {
          bsonType: "array",
          minItems: 1,
          items: { bsonType: "string", minLength: 1, maxLength: 500 }
        },
        data_retention_period_months: { bsonType: ["int", "long"], minimum: 1, maximum: 1200 },
        withdrawal_mechanism: {
          bsonType: "object",
          required: ["sla_remove_external_days", "sla_remove_internal", "audit_log_retention"],
          additionalProperties: false,
          properties: {
            sla_remove_external_days: { bsonType: ["int", "long"], minimum: 0, maximum: 365 },
            sla_remove_internal:      { enum: ["immediate", "scheduled"] },
            audit_log_retention:      { enum: ["permanent", "time_limited"] }
          }
        },
        // Optional CARE fields (Stage 1 pragmatic subset; populated when available)
        ethical_framework_source:      { bsonType: ["string", "null"], maxLength: 500 },
        relationship_custodian_name:   { bsonType: ["string", "null"], maxLength: 500 },
        conflict_resolution_mechanism: { bsonType: ["string", "null"], maxLength: 2000 },
        benefit_sharing_mechanism: {
          enum: [null, "direct_revenue", "capacity_building",
                 "knowledge_access", "policy_influence"]
        },
        local_language_primary: { bsonType: ["string", "null"], pattern: "^[a-z]{2}(-[A-Z]{2})?$" },
        revoked:    { bsonType: "bool" },
        revoked_at: { bsonType: ["date", "null"] },
        created_at: { bsonType: "date" }
      }
    }
  },
  validationLevel: "strict",
  validationAction: "error"
});

// ---------------------------------------------------------------------------
// external_release_logs — append-only public audit trail
// ---------------------------------------------------------------------------
target.createCollection("external_release_logs", {
  validator: {
    $jsonSchema: {
      bsonType: "object",
      required: [
        "schema_version", "release_id", "community",
        "requested_by", "legal_basis", "purpose",
        "field_set", "record_count",
        "privacy_method", "privacy_params",
        "quorum_signatures", "consent_record_ids", "released_at"
      ],
      additionalProperties: false,
      properties: {
        _id: {},
        schema_version: { enum: ["1.0.0"] },
        release_id:     { bsonType: "string", pattern: "^rel_[A-Za-z0-9_-]+$" },
        community:      { bsonType: "string", pattern: "^[a-z]{2}-[a-z0-9-]+$" },
        requested_by:   { bsonType: "string", minLength: 1, maxLength: 500 },
        legal_basis: {
          enum: [
            "dsgvo_art_6_1_a_consent",
            "dsgvo_art_6_1_e_public_interest",
            "dsgvo_art_89_research",
            "lgpd_art_7_consent",
            "lgpd_art_11_sensitive_consent"
          ]
        },
        purpose: { bsonType: "string", minLength: 1, maxLength: 2000 },
        field_set: {
          bsonType: "array", minItems: 1, uniqueItems: true,
          items: { bsonType: "string", minLength: 1 }
        },
        record_count: { bsonType: ["int", "long"], minimum: 0 },
        privacy_method: { enum: ["k_anonymity", "differential_privacy"] },
        privacy_params: {
          bsonType: "object",
          additionalProperties: false,
          properties: {
            k:       { bsonType: ["int", "long"], minimum: 1 },
            epsilon: { bsonType: numericTypes, minimum: 0, exclusiveMinimum: true }
          }
        },
        quorum_signatures: {
          bsonType: "array", minItems: 2,
          items: {
            bsonType: "object",
            required: ["role", "signer", "signed_at"],
            additionalProperties: false,
            properties: {
              role:      { bsonType: "string", minLength: 1, maxLength: 100 },
              signer:    { bsonType: "string", minLength: 1, maxLength: 500 },
              signed_at: { bsonType: "date" }
            }
          }
        },
        consent_record_ids: {
          bsonType: "array", minItems: 1, uniqueItems: true,
          items: { bsonType: "string", pattern: "^csr_[A-Za-z0-9_-]+$" }
        },
        released_at: { bsonType: "date" }
      }
    }
  },
  validationLevel: "strict",
  validationAction: "error"
});

// ---------------------------------------------------------------------------
// fisher_identity_map — vault-strict pseudo↔real mapping.
// Stage 1: collection provisioned, no API endpoint populates it. Helpers in
// privacy.py mint pseudo IDs; population flow lives outside this service.
// ---------------------------------------------------------------------------
target.createCollection("fisher_identity_map", {
  validator: {
    $jsonSchema: {
      bsonType: "object",
      required: ["schema_version", "fisher_pseudo_id", "community",
                 "real_identifier", "created_at"],
      additionalProperties: false,
      properties: {
        _id: {},
        schema_version:   { enum: ["1.0.0"] },
        fisher_pseudo_id: { bsonType: "string", pattern: "^psd_[a-f0-9]{16}$" },
        community:        { bsonType: "string", pattern: "^[a-z]{2}-[a-z0-9-]+$" },
        real_identifier:  { bsonType: "string", minLength: 1 },
        created_at:       { bsonType: "date" }
      }
    }
  },
  validationLevel: "strict",
  validationAction: "error"
});

// ---------------------------------------------------------------------------
// fisher_identities — Stage 2 operational consent registry (app-readable).
// Stores phone_hash (sha256) + pseudonym + per-fisher consent_scope. The
// vault map (fisher_identity_map) holds the plain real_identifier; this
// collection never carries it. Dual-write atomicity is via two-phase
// commit in routes/identity.py (Atlas M0 has no multi-doc transactions).
// ---------------------------------------------------------------------------
target.createCollection("fisher_identities", {
  validator: {
    $jsonSchema: {
      bsonType: "object",
      required: [
        "schema_version", "fisher_pseudo_id", "phone_hash", "pseudonym",
        "community", "consent_scope", "consent_record_id",
        "granted_at", "revoked"
      ],
      additionalProperties: false,
      properties: {
        _id: {},
        schema_version:      { enum: ["1.0.0"] },
        fisher_pseudo_id:    { bsonType: "string", pattern: "^psd_[a-f0-9]{16}$" },
        phone_hash:          { bsonType: "string", pattern: "^[a-f0-9]{64}$" },
        pseudonym: {
          bsonType: "string",
          pattern: "^(tatu|siri|garoupa|robalo|tainha|caranguejo|caica|gaivota|baleia|caranha)-(azul|verde|vermelho|amarelo|preto|branco|rosa|cinza)-[0-9]{2}$"
        },
        community:           { bsonType: "string", pattern: "^[a-z]{2}-[a-z0-9-]+$" },
        consent_scope: {
          bsonType: "array",
          minItems: 1,
          uniqueItems: true,
          items: {
            enum: ["info_access", "tk_preservation",
                   "civic_voice", "catch_contribution"]
          }
        },
        consent_record_id:   { bsonType: "string", pattern: "^csr_[A-Za-z0-9_-]+$" },
        custodian_pseudo_id: { bsonType: ["string", "null"], pattern: "^psd_[a-f0-9]{16}$" },
        granted_at:          { bsonType: "date" },
        revoked:             { bsonType: "bool" },
        revoked_at:          { bsonType: ["date", "null"] }
      }
    }
  },
  validationLevel: "strict",
  validationAction: "error"
});

// ---------------------------------------------------------------------------
// audit_log — append-only mutation ledger.
// 1:1 with mutating routes: each successful POST emits exactly one event.
// entity_id pattern is laxly validated here (oneOf the 3 entity-id prefixes);
// strict cross-field check (entity_type ↔ entity_id) lives in Pydantic.
// ---------------------------------------------------------------------------
target.createCollection("audit_log", {
  validator: {
    $jsonSchema: {
      bsonType: "object",
      required: [
        "schema_version", "audit_event_id", "event_type",
        "entity_type", "entity_id", "community", "recorded_at"
      ],
      additionalProperties: false,
      properties: {
        _id: {},
        schema_version: { enum: ["1.0.0"] },
        audit_event_id: { bsonType: "string", pattern: "^aud_[a-f0-9]{16}$" },
        event_type: {
          enum: ["catch_event_recorded", "consent_granted",
                 "consent_revoked", "release_recorded",
                 "tk_event_recorded", "community_voice_event_recorded",
                 "identity_granted", "identity_revoked"]
        },
        entity_type: {
          enum: ["catch_event", "consent_record", "external_release",
                 "tk_event", "community_voice_event", "fisher_identity"]
        },
        entity_id: {
          bsonType: "string",
          oneOf: [
            { pattern: "^cev_[a-f0-9]{16}$" },
            { pattern: "^csr_[A-Za-z0-9_-]+$" },
            { pattern: "^rel_[A-Za-z0-9_-]+$" },
            { pattern: "^tke_[a-f0-9]{16}$" },
            { pattern: "^cmv_[a-f0-9]{16}$" },
            { pattern: "^psd_[a-f0-9]{16}$" }
          ]
        },
        community:   { bsonType: "string", pattern: "^[a-z]{2}-[a-z0-9-]+$" },
        recorded_at: { bsonType: "date" }
      }
    }
  },
  validationLevel: "strict",
  validationAction: "error"
});

// ---------------------------------------------------------------------------
// traditional_knowledge_events — Stage 1.5 knowledge custodian recordings.
// Append-only. Custodian identity reuses the fisher_identity_map registry
// (same psd_<16hex> pattern); `custodian_pseudo_id` is a contextual label.
// ---------------------------------------------------------------------------
target.createCollection("traditional_knowledge_events", {
  validator: {
    $jsonSchema: {
      bsonType: "object",
      required: [
        "schema_version", "tk_event_id", "community", "custodian_pseudo_id",
        "event_date", "consent_record_id",
        "knowledge_domain", "medium", "transcription_pseudonymized",
        "language_primary", "created_at"
      ],
      additionalProperties: false,
      properties: {
        _id: {},
        schema_version:               { enum: ["1.0.0"] },
        tk_event_id:                  { bsonType: "string", pattern: "^tke_[a-f0-9]{16}$" },
        community:                    { bsonType: "string", pattern: "^[a-z]{2}-[a-z0-9-]+$" },
        custodian_pseudo_id:          { bsonType: "string", pattern: "^psd_[a-f0-9]{16}$" },
        event_date:                   { bsonType: "date" },
        consent_record_id:            { bsonType: "string", pattern: "^csr_[A-Za-z0-9_-]+$" },
        knowledge_domain: {
          enum: ["technique", "species_lore", "calendar", "place_name", "other"]
        },
        medium:                       { enum: ["audio", "text", "photo"] },
        transcription_pseudonymized:  { bsonType: "string", minLength: 1, maxLength: 10000 },
        language_primary:             { bsonType: "string", pattern: "^[a-z]{2}(-[A-Z]{2})?$" },
        created_at:                   { bsonType: "date" }
      }
    }
  },
  validationLevel: "strict",
  validationAction: "error"
});

// ---------------------------------------------------------------------------
// community_voice_events — Stage 1.5 civic incident reports.
// Append-only. Reporter identity reuses fisher_pseudo_id (Stage 1.5 piggy-back;
// a future stage will generalize to informant_pseudo_id for non-fisher
// reporters). PublicView strips fisher_pseudo_id; only vault-role GET exposes it.
// ---------------------------------------------------------------------------
target.createCollection("community_voice_events", {
  validator: {
    $jsonSchema: {
      bsonType: "object",
      required: [
        "schema_version", "community_voice_event_id", "community",
        "fisher_pseudo_id", "event_date", "consent_record_id",
        "incident_type", "location_sector", "severity",
        "evidence_pseudonymized", "escalation_optin", "created_at"
      ],
      additionalProperties: false,
      properties: {
        _id: {},
        schema_version:            { enum: ["1.0.0"] },
        community_voice_event_id:  { bsonType: "string", pattern: "^cmv_[a-f0-9]{16}$" },
        community:                 { bsonType: "string", pattern: "^[a-z]{2}-[a-z0-9-]+$" },
        fisher_pseudo_id:          { bsonType: "string", pattern: "^psd_[a-f0-9]{16}$" },
        event_date:                { bsonType: "date" },
        consent_record_id:         { bsonType: "string", pattern: "^csr_[A-Za-z0-9_-]+$" },
        incident_type: {
          enum: ["illegal_fishing", "oil_spill", "unauthorized_construction",
                 "unauthorized_extraction", "other"]
        },
        location_sector:           { bsonType: "string", minLength: 1, maxLength: 500 },
        severity:                  { enum: ["low", "medium", "high", "critical"] },
        evidence_pseudonymized:    { bsonType: "string", minLength: 1, maxLength: 10000 },
        escalation_optin:          { bsonType: "bool" },
        escalation_target:         { bsonType: ["string", "null"], minLength: 1, maxLength: 500 },
        created_at:                { bsonType: "date" }
      }
    }
  },
  validationLevel: "strict",
  validationAction: "error"
});

// ---------------------------------------------------------------------------
// Indexes
// ---------------------------------------------------------------------------
target.catch_events.createIndex({ catch_event_id: 1 }, { unique: true });
target.catch_events.createIndex({ community: 1, event_date: -1 });
target.catch_events.createIndex({ fisher_pseudo_id: 1, event_date: -1 });
target.catch_events.createIndex({ consent_record_id: 1 });
// Idempotency: same fisher cannot record duplicate catch on same instant for same species.
target.catch_events.createIndex(
  { community: 1, fisher_pseudo_id: 1, event_date: 1, catch_species: 1 },
  { unique: true, name: "uq_catch_event_natural_key" }
);

target.consent_records.createIndex({ consent_record_id: 1 }, { unique: true });
target.consent_records.createIndex({ community: 1, revoked: 1 });

target.external_release_logs.createIndex({ release_id: 1 }, { unique: true });
target.external_release_logs.createIndex({ community: 1, released_at: -1 });

target.fisher_identity_map.createIndex({ fisher_pseudo_id: 1 }, { unique: true });
target.fisher_identity_map.createIndex({ community: 1 });

// fisher_identities — uniqueness partial-filtered on revoked=false so a
// fisher who revokes can re-register with a new pseudo_id/pseudonym.
target.fisher_identities.createIndex(
  { phone_hash: 1 },
  { unique: true,
    partialFilterExpression: { revoked: false },
    name: "uq_identity_phone_hash_active" }
);
target.fisher_identities.createIndex(
  { fisher_pseudo_id: 1 },
  { unique: true,
    partialFilterExpression: { revoked: false },
    name: "uq_identity_pseudo_id_active" }
);
target.fisher_identities.createIndex(
  { community: 1, pseudonym: 1 },
  { unique: true,
    partialFilterExpression: { revoked: false },
    name: "uq_identity_pseudonym_per_community_active" }
);
target.fisher_identities.createIndex(
  { community: 1, granted_at: -1 },
  { name: "ix_identity_community_granted_at" }
);

target.audit_log.createIndex({ audit_event_id: 1 }, { unique: true });
target.audit_log.createIndex({ entity_type: 1, entity_id: 1, recorded_at: -1 });
target.audit_log.createIndex({ community: 1, recorded_at: -1 });
target.audit_log.createIndex({ event_type: 1, recorded_at: -1 });

target.traditional_knowledge_events.createIndex({ tk_event_id: 1 }, { unique: true });
target.traditional_knowledge_events.createIndex({ community: 1, event_date: -1 });
target.traditional_knowledge_events.createIndex({ custodian_pseudo_id: 1, event_date: -1 });
target.traditional_knowledge_events.createIndex({ consent_record_id: 1 });
target.traditional_knowledge_events.createIndex(
  { community: 1, custodian_pseudo_id: 1, event_date: 1, knowledge_domain: 1 },
  { unique: true, name: "uq_tk_event_natural_key" }
);

target.community_voice_events.createIndex({ community_voice_event_id: 1 }, { unique: true });
target.community_voice_events.createIndex({ community: 1, event_date: -1 });
target.community_voice_events.createIndex({ fisher_pseudo_id: 1, event_date: -1 });
target.community_voice_events.createIndex({ consent_record_id: 1 });
target.community_voice_events.createIndex(
  { community: 1, fisher_pseudo_id: 1, event_date: 1, incident_type: 1 },
  { unique: true, name: "uq_community_voice_event_natural_key" }
);

// ---------------------------------------------------------------------------
// Roles + users (defense-in-depth on top of Pydantic-side authorization)
// ---------------------------------------------------------------------------
target.createRole({
  role: "benthosAppRole",
  privileges: [
    // catch_events: append-only.
    { resource: { db: dbName, collection: "catch_events" },
      actions: ["find", "insert", "createIndex", "listIndexes", "listCollections"] },
    // external_release_logs: append-only public log.
    { resource: { db: dbName, collection: "external_release_logs" },
      actions: ["find", "insert", "createIndex", "listIndexes", "listCollections"] },
    // consent_records: read-only — needed to verify consent before accepting
    // catch events or external releases. App cannot create or revoke consents.
    { resource: { db: dbName, collection: "consent_records" },
      actions: ["find", "listIndexes", "listCollections"] },
    // audit_log: append-only ledger — emitted by catch and release routes.
    { resource: { db: dbName, collection: "audit_log" },
      actions: ["find", "insert", "createIndex", "listIndexes", "listCollections"] },
    // traditional_knowledge_events: append-only (Stage 1.5).
    { resource: { db: dbName, collection: "traditional_knowledge_events" },
      actions: ["find", "insert", "createIndex", "listIndexes", "listCollections"] },
    // community_voice_events: append-only (Stage 1.5). App role POSTs and serves
    // the public GET (PublicView strips fisher_pseudo_id).
    { resource: { db: dbName, collection: "community_voice_events" },
      actions: ["find", "insert", "createIndex", "listIndexes", "listCollections"] },
    // fisher_identities: find-only (Stage 2). App role serves the whitelist
    // lookup (GET /v1/identities/by-phone-hash) and the public listing.
    // Writes go through vault role (POST /v1/identities + revoke).
    { resource: { db: dbName, collection: "fisher_identities" },
      actions: ["find", "listIndexes", "listCollections"] }
    // fisher_identity_map: NO access for app role. Vault only.
  ],
  roles: []
});

target.createRole({
  role: "benthosVaultRole",
  privileges: [
    { resource: { db: dbName, collection: "consent_records" },
      actions: ["find", "insert", "update", "createIndex", "listIndexes", "listCollections"] },
    { resource: { db: dbName, collection: "fisher_identity_map" },
      actions: ["find", "insert", "update", "createIndex", "listIndexes", "listCollections"] },
    // audit_log: append-only ledger — emitted by consent routes.
    { resource: { db: dbName, collection: "audit_log" },
      actions: ["find", "insert", "createIndex", "listIndexes", "listCollections"] },
    // community_voice_events: find-only (Stage 1.5). Vault never inserts —
    // POST goes through app role. Used by GET /v1/community-voice-events/vault
    // to expose the full event with reporter identity for privileged operators.
    { resource: { db: dbName, collection: "community_voice_events" },
      actions: ["find", "listIndexes", "listCollections"] },
    // fisher_identities: full RW (Stage 2). Phase 1 of dual-write on POST
    // /v1/identities + revoke + vault-only by-pseudonym operator lookup.
    { resource: { db: dbName, collection: "fisher_identities" },
      actions: ["find", "insert", "update", "createIndex", "listIndexes", "listCollections"] }
  ],
  roles: []
});

target.createUser({
  user: "benthos_app",
  pwd: appPassword,
  roles: [{ role: "benthosAppRole", db: dbName }]
});
target.createUser({
  user: "benthos_vault",
  pwd: vaultPassword,
  roles: [{ role: "benthosVaultRole", db: dbName }]
});

print("[benthos] init complete: collections, validators, indexes, roles, users provisioned for db '" + dbName + "'");