7.8 Migrations and models

Loco convention: a migration/ crate with one file per migration. The schema is kept as a plain SQL file (shown after the migration) and executed as raw SQL, because it uses pgvector and row-level security. A second statement adds the checkpoint table.

// migration/src/m20261110_000001_init.rs
use sea_orm_migration::prelude::*;

#[derive(DeriveMigrationName)]
pub struct Migration;

#[async_trait::async_trait]
impl MigrationTrait for Migration {
    async fn up(&self, m: &SchemaManager) -> Result<(), DbErr> {
        let db = m.get_connection();
        db.execute_unprepared(include_str!("../sql/0001_init.sql")).await?;       // schema below
        db.execute_unprepared(
            "CREATE TABLE workflow_sessions (
               id text PRIMARY KEY,
               tenant_id uuid,
               graph_id text NOT NULL,
               current_task_id text NOT NULL,
               data jsonb NOT NULL,
               version bigint NOT NULL DEFAULT 1,
               updated_at timestamptz NOT NULL DEFAULT now());
             CREATE INDEX workflow_sessions_tenant ON workflow_sessions (tenant_id, updated_at DESC);",
        ).await?;
        Ok(())
    }
    async fn down(&self, m: &SchemaManager) -> Result<(), DbErr> {
        m.get_connection().execute_unprepared("DROP TABLE IF EXISTS workflow_sessions").await?;
        Ok(())
    }
}

-- migration/sql/0001_init.sql
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pgcrypto;

CREATE TABLE tenants (id uuid PRIMARY KEY DEFAULT gen_random_uuid(), slug text NOT NULL UNIQUE, name text NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now());
CREATE TABLE users (id uuid PRIMARY KEY DEFAULT gen_random_uuid(), tenant_id uuid NOT NULL REFERENCES tenants(id),
  email text NOT NULL, roles text[] NOT NULL DEFAULT '{viewer}', created_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (tenant_id, email));
CREATE TABLE vendors (id uuid PRIMARY KEY DEFAULT gen_random_uuid(), tenant_id uuid NOT NULL REFERENCES tenants(id),
  name text NOT NULL, tax_id text, iban_hash text, external_id text, created_at timestamptz NOT NULL DEFAULT now());
CREATE INDEX vendors_tenant_name ON vendors (tenant_id, lower(name));
CREATE TABLE purchase_orders (id uuid PRIMARY KEY DEFAULT gen_random_uuid(), tenant_id uuid NOT NULL REFERENCES tenants(id),
  vendor_id uuid REFERENCES vendors(id), po_number text NOT NULL, total numeric(14,2), currency char(3),
  status text NOT NULL DEFAULT 'open', UNIQUE (tenant_id, po_number));
CREATE TABLE documents (id uuid PRIMARY KEY DEFAULT gen_random_uuid(), tenant_id uuid NOT NULL REFERENCES tenants(id),
  filename text NOT NULL, sha256 bytea NOT NULL, mime text, storage_uri text NOT NULL,
  parsed_md text, parsed_json jsonb, status text NOT NULL DEFAULT 'received', created_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (tenant_id, sha256));                                   -- idempotent ingest
CREATE TABLE extracted_invoices (id uuid PRIMARY KEY DEFAULT gen_random_uuid(), tenant_id uuid NOT NULL REFERENCES tenants(id),
  document_id uuid NOT NULL REFERENCES documents(id), vendor_id uuid REFERENCES vendors(id), po_id uuid REFERENCES purchase_orders(id),
  invoice_number text NOT NULL, invoice_date date, due_date date, currency char(3),
  subtotal numeric(14,2), tax numeric(14,2), total numeric(14,2),
  data jsonb NOT NULL, validation jsonb NOT NULL DEFAULT '{}', match_score real,
  status text NOT NULL DEFAULT 'extracted', external_id text, model_id text NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (tenant_id, vendor_id, invoice_number));                -- duplicate-invoice guard
CREATE TABLE workflow_runs (id uuid PRIMARY KEY DEFAULT gen_random_uuid(), tenant_id uuid NOT NULL REFERENCES tenants(id),
  workflow_id text NOT NULL, workflow_version int NOT NULL, session_id text NOT NULL UNIQUE, status text NOT NULL,
  started_by uuid REFERENCES users(id), created_at timestamptz NOT NULL DEFAULT now(), updated_at timestamptz NOT NULL DEFAULT now());
CREATE TABLE approvals (id uuid PRIMARY KEY DEFAULT gen_random_uuid(), tenant_id uuid NOT NULL REFERENCES tenants(id),
  session_id text NOT NULL REFERENCES workflow_runs(session_id), interrupt_id text NOT NULL UNIQUE,
  summary jsonb NOT NULL, status text NOT NULL DEFAULT 'pending', decision jsonb, decided_by uuid REFERENCES users(id),
  expires_at timestamptz, decided_at timestamptz, created_at timestamptz NOT NULL DEFAULT now());
CREATE TABLE audit_log (id bigserial PRIMARY KEY, tenant_id uuid NOT NULL, actor text NOT NULL, action text NOT NULL,
  subject text NOT NULL, data jsonb NOT NULL, trace_id text, prev_hash bytea, hash bytea NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now());                -- hash-chained, append-only
CREATE INDEX audit_tenant_time ON audit_log (tenant_id, created_at DESC);
REVOKE UPDATE, DELETE ON audit_log FROM PUBLIC;
CREATE TABLE idempotency_keys (key text PRIMARY KEY, tenant_id uuid NOT NULL, response jsonb,
  created_at timestamptz NOT NULL DEFAULT now());
CREATE TABLE embeddings (id bigserial PRIMARY KEY, tenant_id uuid NOT NULL,
  document_id uuid NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
  chunk_ix int NOT NULL, content text NOT NULL, embedding vector(1024) NOT NULL, model_id text NOT NULL);
CREATE INDEX embeddings_hnsw ON embeddings USING hnsw (embedding vector_cosine_ops);
CREATE INDEX embeddings_tenant ON embeddings (tenant_id);

-- Row-level security as a second layer behind the application's tenant filters.
-- The app sets `app.tenant_id` per transaction: SELECT set_config('app.tenant_id', $1, true).
DO $$ DECLARE t text; BEGIN
  FOREACH t IN ARRAY ARRAY['users','vendors','purchase_orders','documents','extracted_invoices',
                           'workflow_runs','approvals','audit_log','embeddings'] LOOP
    EXECUTE format('ALTER TABLE %I ENABLE ROW LEVEL SECURITY', t);
    EXECUTE format('CREATE POLICY tenant_isolation ON %I USING (tenant_id = current_setting(''app.tenant_id'')::uuid)', t);
  END LOOP; END $$;

The embedding column width (1024) must match dimensions for the embedding model in models.yaml. Changing the embedding model to a different width needs a migration.

• Entities are generated into app/src/models/_entities/ with bo-app db entities (VERIFY command name on 1.3).
• Tenant scoping uses Loco's helpers: implement TenantEntity per entity, then Entity::find().in_tenant(tenant_id) and active_model.set_tenant(tenant_id)? (source-verified). Postgres row-level security stays as a second layer.
• Vector search uses a raw statement, because the pgvector crate has no SeaORM integration:

// app/src/models/embeddings.rs
use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, Statement};

pub async fn search(db: &DatabaseConnection, tenant: uuid::Uuid, query: &[f32], k: i64)
    -> Result<Vec<(uuid::Uuid, String, f64)>, sea_orm::DbErr> {
    let literal = format!("[{}]", query.iter().map(f32::to_string).collect::<Vec<_>>().join(","));
    let rows = db.query_all_raw(Statement::from_sql_and_values(DbBackend::Postgres,     // VERIFY name on SeaORM 2.0
        "SELECT document_id, content, 1 - (embedding <=> $2::vector) AS score
           FROM embeddings WHERE tenant_id = $1
          ORDER BY embedding <=> $2::vector LIMIT $3",
        [tenant.into(), literal.into(), k.into()])).await?;
    rows.into_iter().map(|r| Ok((r.try_get("", "document_id")?, r.try_get("", "content")?, r.try_get("", "score")?))).collect()
}