"""Task 3: the three fiscal tables that "change owner" per the design spec (decision #2) -- they used to live in the auto (`app/modules/tenants/ models.py`'s `FiscalDocumentSeries`, `app/modules/fiscal/models.py`'s `FiscalCertificate`/`FiscalDocument`) and now live HERE, as the source of truth this service was built to be. Ported with the porte table (2026-07-17-fiscal-svc-f2-servico.md) applied throughout: | No auto | No serviço | |-----------------------------------|-----------------------------------| | organization_id (FK organizations)| product_id (FK products) + | | | tenant_ref: String(64) | | branch_id (FK branches) | branch_ref: String(64) + cnpj | | | (the strong link) | | Sale/rotas por venda | não existem (sale_id/ | | | service_order_id DROPPED -- | | | emissão recebe DadosEmissao | | | completo, sem correlação a uma | | | linha de venda deste lado) | Global Constraints (zero enum PG): unlike the auto's `FiscalDocumentSeries. document_model` (a REAL Postgres `Enum` -- a documented, deliberately un-repeated debt there), `document_model` here is plain `String(2)` + Python enum at the Pydantic border ONLY, same convention the auto's own `FiscalDocument.document_model`/`.status` already used (see their docstrings) -- this port does not re-introduce the PG enum anywhere.""" import enum import uuid from datetime import datetime from sqlalchemy import ( DateTime, ForeignKey, Index, Integer, LargeBinary, String, Text, UniqueConstraint, text, ) from sqlalchemy.orm import Mapped, mapped_column from fiscal_svc.core.db import Base from fiscal_svc.shared.base_model import SoftDeleteMixin, TimestampMixin, UUIDPKMixin class FiscalDocumentModel(str, enum.Enum): """Ported from `auto/backend/app/modules/tenants/models.py`. Python enum only -- see this module's docstring for why the auto's Postgres `Enum` column for this same enum is NOT repeated here.""" NFE_55 = "55" NFCE_65 = "65" class FiscalDocumentStatus(str, enum.Enum): """Ported verbatim from `auto/backend/app/modules/fiscal/models.py` -- already `String` + Python enum there (ZERO Postgres enum), so no adaptation needed. State machine: `ASSINADO -> TRANSMITINDO -> AUTORIZADA | REJEITADA | PENDENTE_CONSULTA` (+ `CANCELADA`/`DENEGADA`). Task 3 (this port) only ever WRITES `ASSINADO` -- transmission (F4) writes the rest; the full vocabulary exists now so a later phase never needs a column-shape migration for a new status.""" ASSINADO = "ASSINADO" TRANSMITINDO = "TRANSMITINDO" AUTORIZADA = "AUTORIZADA" REJEITADA = "REJEITADA" PENDENTE_CONSULTA = "PENDENTE_CONSULTA" CANCELADA = "CANCELADA" DENEGADA = "DENEGADA" class FiscalSeries(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin): """Ported from the auto's `FiscalDocumentSeries` (`tenants/models.py`) -- per the design spec this table is the NEW owner of fiscal numbering (decision #2: "mudam de dono"). `product_id`/`tenant_ref`/`branch_ref` replace `organization_id`/`branch_id` per the porte table (this module's docstring). Atomic allocation (`SELECT ... FOR UPDATE` + `populate_existing`) is `documents.service.allocate_fiscal_number` (Task 3, ported with its exact no-commit contract -- see that function's docstring). The "guard retroativo" (`next_number` can never regress below the highest `numero` this series has already emitted, per `FiscalDocument.numero`) is deliberately NOT implemented in this task -- it belongs to the `PATCH /v1/series` endpoint (Task 4), which needs this exact model plus `FiscalDocument` (both now exist, Task 3) to check against. Same "documented now, wired later" shape as the auto's own `FiscalDocumentSeries` docstring ("a emissão em si nasce no módulo fiscal da 1b").""" __tablename__ = "fiscal_series" __table_args__ = ( UniqueConstraint( "product_id", "tenant_ref", "branch_ref", "document_model", "serie", name="uq_fiscal_series_product_tenant_branch_model_serie", ), ) product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False) tenant_ref: Mapped[str] = mapped_column(String(64), nullable=False) branch_ref: Mapped[str] = mapped_column(String(64), nullable=False) document_model: Mapped[str] = mapped_column(String(2), nullable=False) serie: Mapped[int] = mapped_column(Integer, nullable=False) next_number: Mapped[int] = mapped_column(Integer, nullable=False) class FiscalCertificate(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin): """Ported from the auto's `FiscalCertificate` (`app/modules/fiscal/ models.py`) -- A1 certificate (.pfx) for a `branch_ref`. Per the porte table, the emitter is `(product_id, tenant_ref, branch_ref)`, with `cnpj` carried on the row itself (the "vínculo forte", design spec decision #4: the CNPJ is what's validated against the certificate at upload and against the emitente at emission -- `branch_ref` alone is an opaque string this service never interprets). No fallback to a product-level certificate -- fail-closed by construction, same as the auto: a `(tenant_ref, branch_ref)` without its own live certificate cannot emit. One VIVO (`deleted_at IS NULL`) row per `(product_id, tenant_ref, branch_ref)`: a second upload soft-deletes the previous live row (Task 4's `certificates.service`). `ix_fiscal_certificates_product_ tenant_branch_live` is a PARTIAL UNIQUE index enforcing that at the database level (same "two concurrent uploads must not both land a live row" reasoning as the auto's own migration `a1b2c3d4e5f6`'s fix) -- scoped by `product_id` AND `tenant_ref` in ADDITION to `branch_ref` (the auto's version only needed `branch_id`, already product-scoped by being a real FK; here `branch_ref` is an OPAQUE string owned by the calling product, so two DIFFERENT products -- or two DIFFERENT tenants of the SAME product -- could coincidentally pick the identical string for two DIFFERENT real branches -- scoping the uniqueness by `product_id`+`tenant_ref` too is what keeps that from cross-contaminating one tenant's certificate slot with another's; FIX 2, F2 review, 2026-07-17-sowai-fiscal-svc-design.md decisão #4 turned this into a shipped bug otherwise: two tenants both using `branch_ref= "matriz"` would collapse onto one slot, the second tenant's upload silently soft-deleting the first's still-live certificate). `pfx_encrypted`/`password_encrypted` are Fernet ciphertext (Task 4, `FISCAL_CERT_ENCRYPTION_KEY` env var, own key -- never `app.core.config.Settings`, same "secrets never live in Settings" convention as the auto).""" __tablename__ = "fiscal_certificates" __table_args__ = ( Index( "ix_fiscal_certificates_product_tenant_branch_live", "product_id", "tenant_ref", "branch_ref", unique=True, postgresql_where=text("deleted_at IS NULL"), ), ) product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False) tenant_ref: Mapped[str] = mapped_column(String(64), nullable=False) branch_ref: Mapped[str] = mapped_column(String(64), nullable=False) cnpj: Mapped[str] = mapped_column(String(14), nullable=False) pfx_encrypted: Mapped[bytes] = mapped_column(LargeBinary, nullable=False) password_encrypted: Mapped[bytes] = mapped_column(LargeBinary, nullable=False) subject_cn: Mapped[str] = mapped_column(String(255), nullable=False) cnpj_certificado: Mapped[str] = mapped_column(String(14), nullable=False) not_valid_before: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) not_valid_after: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) class FiscalDocument(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin): """Ported from the auto's `FiscalDocument` (`app/modules/fiscal/ models.py`). Per the porte table, `sale_id`/`service_order_id` are DROPPED -- this service has no notion of a Sale/ServiceOrder (decision #3 of the design spec: it receives a complete `DadosEmissao`, never resolves or references the caller's own domain objects). A product's own correlation to ITS sale/OS is the product's problem, carried via the `Idempotency-Key` (Task 5) it sends, not a column here. Invariants preserved from the auto (contract, not mere convention): (a) a row is only inserted in the SAME commit that allocates `numero` via `documents.service.allocate_fiscal_number` (which does not commit by itself -- see its docstring) -- outbox transactional, exactly as ported; (b) documento é IMUTÁVEL após `ASSINADO` -- correção é cancelar (evento futuro) e emitir outro, nunca um UPDATE no XML/chave já assinados; (c) `chave_acesso` is UNIQUE GLOBALLY (not just per product) -- it already embeds the emitente's CNPJ in its own digits, so two DIFFERENT products (or tenants) can never legitimately produce the same one; a collision here can only be a generation bug, never a false positive between tenants OR between products. `ix_fiscal_documents_product_tenant_branch_status` replaces the auto's `ix_fiscal_documents_org_branch_status` (porte table: anti-oracle 404 is now by `(product_id, tenant_ref)`, and `GET /v1/documentos` filters by `tenant_ref`/`branch_ref`/`status` per the design spec's API table) -- the index covers exactly that filter shape.""" __tablename__ = "fiscal_documents" __table_args__ = ( Index( "ix_fiscal_documents_product_tenant_branch_status", "product_id", "tenant_ref", "branch_ref", "status", ), UniqueConstraint("chave_acesso", name="uq_fiscal_documents_chave_acesso"), ) product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False) tenant_ref: Mapped[str] = mapped_column(String(64), nullable=False) branch_ref: Mapped[str] = mapped_column(String(64), nullable=False) series_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("fiscal_series.id"), nullable=False) document_model: Mapped[str] = mapped_column(String(2), nullable=False) serie: Mapped[int] = mapped_column(Integer, nullable=False) numero: Mapped[int] = mapped_column(Integer, nullable=False) chave_acesso: Mapped[str] = mapped_column(String(44), nullable=False) codigo_numerico: Mapped[str] = mapped_column(String(8), nullable=False) status: Mapped[str] = mapped_column(String(20), nullable=False) ambiente: Mapped[str] = mapped_column(String(12), nullable=False) xml_assinado: Mapped[str] = mapped_column(Text, nullable=False) rejeicao_codigo: Mapped[str | None] = mapped_column(String(10), nullable=True) rejeicao_motivo: Mapped[str | None] = mapped_column(String(500), nullable=True) protocolo: Mapped[str | None] = mapped_column(String(20), nullable=True) autorizada_em: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) class FiscalIdempotencyKey(Base, UUIDPKMixin, TimestampMixin): """Task 5: backs the `Idempotency-Key` contract of `POST /v1/emissoes` (design spec decision #6) -- a SEPARATE table rather than a column on `FiscalDocument` (Task 3's table, already shipped/migrated) so this Task never touches that migration. `UNIQUE (product_id, idempotency_key)` is the SAME-transaction outbox partner of `emission. service.emitir_documento`'s number-allocation + document-INSERT commit (`documents.service.allocate_fiscal_number`'s docstring): this row is added to the SAME session, in the SAME commit, as the `FiscalDocument` it points to via `document_id` -- so a rollback (e.g. the idempotency race below, or a signature failure) undoes the allocated number AND the document row AND this key together, never just some of the three. The two-layer idempotency pattern this table exists for: (1) a cheap pre-check SELECT before doing any real work (the fast path for a genuine retry); (2) this UNIQUE constraint as the source of truth for the RACE -- two concurrent requests carrying the SAME `Idempotency-Key` can both pass the pre-check (`None`) before either commits; the FIRST to commit wins, the SECOND's commit raises `IntegrityError` here, which `emission.service` catches and translates into a RE-READ of the winner's row (converging both callers on the SAME `FiscalDocument`, never emitting a duplicate NF-e for one logical request). No soft-delete mixin: an idempotency key's history is permanent by design -- there is no "un-claim this key" operation.""" __tablename__ = "fiscal_idempotency_keys" __table_args__ = ( UniqueConstraint( "product_id", "idempotency_key", name="uq_fiscal_idempotency_key_product_key", ), ) product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False) idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False) document_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("fiscal_documents.id"), nullable=False)