"""M1 任务域：item / tag / item_tag / comment / activity + 行级安全。 RLS 策略与 0001 同一套模式：`workspace_id = app_ws_id()`，读写都校验。 `item` 额外允许"我创建或分配给我"——不，**不做这个特例**：跨租户看别人的任务没有业务含义， 越权面越小越好，列表查询本来就带 workspace_id 过滤。 注意：`pg_trgm` 是 trusted extension（PG13+），库 owner 就能装，所以迁移不需要 superuser。 """ from __future__ import annotations from collections.abc import Sequence import sqlalchemy as sa from alembic import op from sqlalchemy.dialects import postgresql revision: str = "0002" down_revision: str | None = "0001" branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None APP_ROLE = "dadao_app" TENANT_TABLES_0002: tuple[str, ...] = ("item", "tag", "item_tag", "comment", "activity") _TIMESTAMP_COLUMNS = ( sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("version", sa.Integer, nullable=False, server_default="1"), ) def _ws_fk() -> sa.Column: return sa.Column( "workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False, ) def upgrade() -> None: # 中文没有分词器（zhparser 未装），M1 的搜索走 ILIKE 子串匹配， # trigram 索引能让"标题里找关键字"这条最常用的查询走索引而不是全表扫。 op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm") op.create_table( "item", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), _ws_fk(), sa.Column("kind", sa.String(16), nullable=False, server_default="task"), sa.Column("title", sa.String(200), nullable=False), sa.Column("body_md", sa.Text, nullable=False, server_default=""), sa.Column("status", sa.String(16), nullable=False, server_default="inbox"), sa.Column("priority", sa.Integer, nullable=False, server_default="0"), sa.Column( "creator_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("app_user.id", ondelete="RESTRICT"), nullable=False, ), sa.Column( "assignee_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("app_user.id", ondelete="SET NULL"), ), sa.Column("due_at", sa.DateTime(timezone=True)), sa.Column("completed_at", sa.DateTime(timezone=True)), sa.Column("sort_order", sa.Integer, nullable=False, server_default="0"), sa.Column("deleted_at", sa.DateTime(timezone=True)), *_TIMESTAMP_COLUMNS, sa.CheckConstraint( "status IN ('inbox','todo','doing','done','blocked','canceled')", name="ck_item_status", ), sa.CheckConstraint("kind IN ('task','ticket','knowledge')", name="ck_item_kind"), sa.CheckConstraint("priority BETWEEN 0 AND 3", name="ck_item_priority"), ) op.create_index("ix_item_ws_status_due", "item", ["workspace_id", "status", "due_at"]) op.create_index("ix_item_ws_assignee", "item", ["workspace_id", "assignee_id", "status"]) op.create_index("ix_item_ws_updated", "item", ["workspace_id", "updated_at"]) op.create_index("ix_item_ws_sort", "item", ["workspace_id", "status", "sort_order"]) op.create_table( "tag", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), _ws_fk(), sa.Column("name", sa.String(32), nullable=False), sa.Column("color", sa.String(16), nullable=False, server_default=""), *_TIMESTAMP_COLUMNS, sa.UniqueConstraint("workspace_id", "name", name="uq_tag_ws_name"), ) op.create_index("ix_tag_ws", "tag", ["workspace_id", "name"]) op.create_table( "item_tag", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), _ws_fk(), sa.Column( "item_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("item.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "tag_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("tag.id", ondelete="CASCADE"), nullable=False, ), *_TIMESTAMP_COLUMNS, sa.UniqueConstraint("item_id", "tag_id", name="uq_item_tag"), ) op.create_index("ix_item_tag_ws_tag", "item_tag", ["workspace_id", "tag_id"]) op.create_table( "comment", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), _ws_fk(), sa.Column( "item_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("item.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "author_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("app_user.id", ondelete="RESTRICT"), nullable=False, ), sa.Column("body_md", sa.Text, nullable=False, server_default=""), sa.Column("deleted_at", sa.DateTime(timezone=True)), *_TIMESTAMP_COLUMNS, ) op.create_index("ix_comment_ws_item_created", "comment", ["workspace_id", "item_id", "created_at"]) op.create_table( "activity", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), _ws_fk(), sa.Column( "item_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("item.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "actor_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("app_user.id", ondelete="SET NULL"), ), sa.Column("action", sa.String(32), nullable=False), sa.Column("payload", postgresql.JSONB), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), ) op.create_index("ix_activity_ws_item_created", "activity", ["workspace_id", "item_id", "created_at"]) # 标题三元组索引：ILIKE '%关键字%' 能用到它 op.execute( "CREATE INDEX ix_item_title_trgm ON item USING gin (title gin_trgm_ops)" ) for statement in _POLICY_STATEMENTS: op.execute(statement) for statement in _GRANT_STATEMENTS: op.execute(statement) def downgrade() -> None: # 先摘策略：策略里引用了这些表本身，表没了策略语句就悬空 for table in TENANT_TABLES_0002: op.execute(f"DROP POLICY IF EXISTS {table}_isolation ON {table}") op.execute("DROP INDEX IF EXISTS ix_item_title_trgm") for table in ( "activity", "comment", "item_tag", "tag", "item", ): op.drop_table(table) # pg_trgm 是库级共享资源，别在这里删（其它模块可能也在用） _POLICY_STATEMENTS: tuple[str, ...] = tuple( statement for table in TENANT_TABLES_0002 for statement in ( f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY;", f""" CREATE POLICY {table}_isolation ON {table} USING (workspace_id = app_ws_id()) WITH CHECK (workspace_id = app_ws_id()); """, ) ) _GRANT_STATEMENTS: tuple[str, ...] = ( f"GRANT SELECT, INSERT, UPDATE, DELETE ON {', '.join(TENANT_TABLES_0002)} TO {APP_ROLE};", )