feat(email): deterministic dedupe provenance — gate strong fingerprints on genuine Date (naruon#1086)

backend/alembic/versions/0018_email_date_provenance.py

@@ -0,0 +1,49 @@
+"""add date_provenance to email_records
+
+Revision ID: 0018_email_date_provenance
+Revises: 0017_merge_newsdom_carddav_heads
+Create Date: 2026-07-30 00:00:00.000000
+
+Records the provenance of each stored email ``date`` so a synthetic
+collection-time fallback (missing/invalid RFC822 Date header) is never treated
+as original sender metadata when seeding a strong auto-dedupe fingerprint
+(naruon#1086). Nullable-free with a ``"unknown"`` server default so existing
+rows backfill safely: their date provenance is genuinely unknown, and only
+``"parsed"`` rows are eligible to seed a strong fingerprint, so the backfill is
+conservative (it can only widen review, never manufacture a duplicate).
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0018_email_date_provenance"
+down_revision = "0017_merge_newsdom_carddav_heads"
+
+_EMAIL_TABLE = "email_records"
+_PROVENANCE_COLUMN = "date_provenance"
+
+
+def upgrade() -> None:
+    """Add the ``date_provenance`` column, backfilling existing rows to unknown."""
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+    columns = {column["name"] for column in inspector.get_columns(_EMAIL_TABLE)}
+    if _PROVENANCE_COLUMN not in columns:
+        op.add_column(
+            _EMAIL_TABLE,
+            sa.Column(
+                _PROVENANCE_COLUMN,
+                sa.String(),
+                nullable=False,
+                server_default="unknown",
+            ),
+        )
+
+
+def downgrade() -> None:
+    """Drop the ``date_provenance`` column if present."""
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+    columns = {column["name"] for column in inspector.get_columns(_EMAIL_TABLE)}
+    if _PROVENANCE_COLUMN in columns:
+        op.drop_column(_EMAIL_TABLE, _PROVENANCE_COLUMN)

backend/alembic/versions/0019_pop3_observed_uidl.py

@@ -0,0 +1,76 @@
+"""add durable POP3 UIDL collection progress
+
+Revision ID: 0019_pop3_observed_uidl
+Revises: 0018_email_date_provenance
+Create Date: 2026-09-17 00:00:00.000000
+
+Persists RFC 1939 UIDL provider identity and bounded retry disposition per
+mailbox configuration. Provider state remains collection progress only; Naruon
+email Message-ID and source fingerprints remain canonical message/deduplication
+evidence.
+
+This revision identifier and parent are branch-local until the canonical
+workspace/Alembic owner is integrated; #1195 must rechain this schema after the
+then-protected migration head before merge.
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0019_pop3_observed_uidl"
+down_revision = "0018_email_date_provenance"
+
+_TABLE = "pop3_observed_messages"
+
+
+def upgrade() -> None:
+    """Create owner-scoped durable POP3 UIDL collection progress if absent."""
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+    if _TABLE in inspector.get_table_names():
+        return
+
+    op.create_table(
+        _TABLE,
+        sa.Column("observed_message_id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "tenant_config_id",
+            sa.Integer(),
+            sa.ForeignKey("tenant_configs.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("provider_uidl", sa.String(length=70), nullable=False),
+        sa.Column(
+            "collection_disposition",
+            sa.String(length=16),
+            nullable=False,
+            server_default="observed",
+        ),
+        sa.Column("retry_after", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("observed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.UniqueConstraint(
+            "tenant_config_id",
+            "provider_uidl",
+            name="uq_pop3_observed_messages_account_uidl",
+        ),
+    )
+    op.create_index(
+        "ix_pop3_observed_messages_account_observed",
+        _TABLE,
+        ["tenant_config_id", "observed_at"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_pop3_observed_messages_account_retry",
+        _TABLE,
+        ["tenant_config_id", "collection_disposition", "retry_after"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    """Drop durable POP3 UIDL collection progress if present."""
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+    if _TABLE in inspector.get_table_names():
+        op.drop_table(_TABLE)

backend/db/models.py

@@ -801,6 +801,13 @@ def owner_filters(cls, user_id: str, organization_id: str | None):
     in_reply_to: Mapped[str | None] = mapped_column(String, nullable=True)
     references: Mapped[str | None] = mapped_column(String, nullable=True)
     date: Mapped[datetime.datetime] = mapped_column(DateTime(timezone=True), index=True)
+    # Provenance of the stored ``date``: "parsed" (genuine RFC822 Date header),
+    # "missing"/"invalid" (a synthetic collection-time fallback), or "unknown"
+    # (rows stored before provenance tracking). Only "parsed" rows may seed a
+    # strong auto-dedupe fingerprint (naruon#1086).
+    date_provenance: Mapped[str] = mapped_column(
+        String, nullable=False, server_default="unknown", default="unknown"
+    )
     body: Mapped[str] = mapped_column(Text)
     # IMAP \Seen read state; defaults read so historical/file imports don't nag.
     is_read: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)

backend/db/pop3_collection_models.py

@@ -0,0 +1,58 @@
+import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Index, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from db.models import Base
+
+
+class Pop3ObservedMessage(Base):
+    """Durable provider progress for one POP3 UIDL within an account.
+
+    POP3 message numbers are session-local positions. ``provider_uidl`` stores
+    RFC 1939 provider identity so bounded polling can make progress across
+    reconnects and renumbering. ``collection_disposition`` distinguishes
+    successfully observed messages from retryable collection failures; neither
+    state redefines Naruon's Message-ID or source-fingerprint identity.
+    """
+
+    __tablename__ = "pop3_observed_messages"
+    __table_args__ = (
+        UniqueConstraint(
+            "tenant_config_id",
+            "provider_uidl",
+            name="uq_pop3_observed_messages_account_uidl",
+        ),
+        Index(
+            "ix_pop3_observed_messages_account_observed",
+            "tenant_config_id",
+            "observed_at",
+        ),
+        Index(
+            "ix_pop3_observed_messages_account_retry",
+            "tenant_config_id",
+            "collection_disposition",
+            "retry_after",
+        ),
+    )
+
+    observed_message_id: Mapped[int] = mapped_column(primary_key=True)
+    tenant_config_id: Mapped[int] = mapped_column(
+        ForeignKey("tenant_configs.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    provider_uidl: Mapped[str] = mapped_column(String(70), nullable=False)
+    collection_disposition: Mapped[str] = mapped_column(
+        String(16),
+        default="observed",
+        nullable=False,
+    )
+    retry_after: Mapped[datetime.datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+    )
+    observed_at: Mapped[datetime.datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.datetime.now(datetime.timezone.utc),
+        nullable=True,
+    )