804 lines
29 KiB
Python
804 lines
29 KiB
Python
from sqlalchemy import inspect, text
|
|
from sqlmodel import SQLModel, Session, create_engine
|
|
|
|
from core.settings import (
|
|
DATABASE_ECHO,
|
|
DATABASE_ENGINE,
|
|
DATABASE_MAX_OVERFLOW,
|
|
DATABASE_POOL_RECYCLE,
|
|
DATABASE_POOL_SIZE,
|
|
DATABASE_POOL_TIMEOUT,
|
|
DATABASE_URL,
|
|
)
|
|
|
|
# Ensure table models are registered in SQLModel metadata before create_all.
|
|
from models import bot as _bot_models # noqa: F401
|
|
from models import platform as _platform_models # noqa: F401
|
|
from models import skill as _skill_models # noqa: F401
|
|
from models import topic as _topic_models # noqa: F401
|
|
|
|
_engine_kwargs = {
|
|
"echo": DATABASE_ECHO,
|
|
}
|
|
if DATABASE_ENGINE == "sqlite":
|
|
_engine_kwargs["connect_args"] = {"check_same_thread": False}
|
|
else:
|
|
_engine_kwargs.update(
|
|
{
|
|
"pool_pre_ping": True,
|
|
"pool_size": DATABASE_POOL_SIZE,
|
|
"max_overflow": DATABASE_MAX_OVERFLOW,
|
|
"pool_timeout": DATABASE_POOL_TIMEOUT,
|
|
"pool_recycle": DATABASE_POOL_RECYCLE,
|
|
}
|
|
)
|
|
|
|
engine = create_engine(DATABASE_URL, **_engine_kwargs)
|
|
|
|
BOT_INSTANCE_TABLE = "bot_instance"
|
|
BOT_MESSAGE_TABLE = "bot_message"
|
|
BOT_IMAGE_TABLE = "bot_image"
|
|
BOT_REQUEST_USAGE_TABLE = "bot_request_usage"
|
|
BOT_ACTIVITY_EVENT_TABLE = "bot_activity_event"
|
|
SYS_SETTING_TABLE = "sys_setting"
|
|
POSTGRES_MIGRATION_LOCK_KEY = 2026031801
|
|
MYSQL_MIGRATION_LOCK_NAME = "dashboard_nanobot_schema_migration"
|
|
LEGACY_TABLE_PAIRS = [
|
|
("botinstance", BOT_INSTANCE_TABLE),
|
|
("botmessage", BOT_MESSAGE_TABLE),
|
|
("nanobotimage", BOT_IMAGE_TABLE),
|
|
("platformsetting", SYS_SETTING_TABLE),
|
|
("botrequestusage", BOT_REQUEST_USAGE_TABLE),
|
|
("botactivityevent", BOT_ACTIVITY_EVENT_TABLE),
|
|
]
|
|
|
|
|
|
def _quote_ident(name: str) -> str:
|
|
if engine.dialect.name == "mysql":
|
|
return f"`{str(name).replace('`', '``')}`"
|
|
return f'"{str(name).replace(chr(34), chr(34) * 2)}"'
|
|
|
|
|
|
def _rename_table_if_needed(old_name: str, new_name: str) -> None:
|
|
inspector = inspect(engine)
|
|
if not inspector.has_table(old_name) or inspector.has_table(new_name):
|
|
return
|
|
dialect = engine.dialect.name
|
|
with engine.connect() as conn:
|
|
if dialect == "mysql":
|
|
conn.execute(text(f"RENAME TABLE `{old_name}` TO `{new_name}`"))
|
|
else:
|
|
conn.execute(text(f'ALTER TABLE "{old_name}" RENAME TO "{new_name}"'))
|
|
conn.commit()
|
|
|
|
|
|
def _rename_legacy_tables() -> None:
|
|
_rename_table_if_needed("botinstance", BOT_INSTANCE_TABLE)
|
|
_rename_table_if_needed("botmessage", BOT_MESSAGE_TABLE)
|
|
_rename_table_if_needed("nanobotimage", BOT_IMAGE_TABLE)
|
|
_rename_table_if_needed("platformsetting", SYS_SETTING_TABLE)
|
|
_rename_table_if_needed("botrequestusage", BOT_REQUEST_USAGE_TABLE)
|
|
_rename_table_if_needed("botactivityevent", BOT_ACTIVITY_EVENT_TABLE)
|
|
|
|
|
|
def _acquire_migration_lock():
|
|
if engine.dialect.name == "postgresql":
|
|
conn = engine.connect()
|
|
conn.execute(text("SELECT pg_advisory_lock(:key)"), {"key": POSTGRES_MIGRATION_LOCK_KEY})
|
|
return conn
|
|
if engine.dialect.name == "mysql":
|
|
conn = engine.connect()
|
|
acquired = conn.execute(
|
|
text("SELECT GET_LOCK(:name, :timeout)"),
|
|
{"name": MYSQL_MIGRATION_LOCK_NAME, "timeout": 120},
|
|
).scalar()
|
|
if int(acquired or 0) != 1:
|
|
conn.close()
|
|
raise RuntimeError("Failed to acquire schema migration lock")
|
|
return conn
|
|
return None
|
|
|
|
|
|
def _release_migration_lock(lock_conn) -> None:
|
|
if lock_conn is None:
|
|
return
|
|
try:
|
|
if engine.dialect.name == "postgresql":
|
|
lock_conn.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": POSTGRES_MIGRATION_LOCK_KEY})
|
|
elif engine.dialect.name == "mysql":
|
|
lock_conn.execute(text("SELECT RELEASE_LOCK(:name)"), {"name": MYSQL_MIGRATION_LOCK_NAME})
|
|
finally:
|
|
lock_conn.close()
|
|
|
|
|
|
def _table_row_count(table_name: str) -> int:
|
|
inspector = inspect(engine)
|
|
if not inspector.has_table(table_name):
|
|
return 0
|
|
with engine.connect() as conn:
|
|
value = conn.execute(text(f"SELECT COUNT(*) FROM {_quote_ident(table_name)}")).scalar()
|
|
return int(value or 0)
|
|
|
|
|
|
def _copy_legacy_table_rows(old_name: str, new_name: str) -> None:
|
|
inspector = inspect(engine)
|
|
if not inspector.has_table(old_name) or not inspector.has_table(new_name):
|
|
return
|
|
if _table_row_count(old_name) <= 0:
|
|
return
|
|
|
|
old_columns = {
|
|
str(row.get("name"))
|
|
for row in inspector.get_columns(old_name)
|
|
if row.get("name")
|
|
}
|
|
new_columns = [
|
|
str(row.get("name"))
|
|
for row in inspector.get_columns(new_name)
|
|
if row.get("name")
|
|
]
|
|
shared_columns = [col for col in new_columns if col in old_columns]
|
|
if not shared_columns:
|
|
return
|
|
pk = inspector.get_pk_constraint(new_name) or {}
|
|
pk_columns = [
|
|
str(col)
|
|
for col in (pk.get("constrained_columns") or [])
|
|
if col and col in shared_columns and col in old_columns
|
|
]
|
|
if not pk_columns:
|
|
return
|
|
|
|
columns_sql = ", ".join(_quote_ident(col) for col in shared_columns)
|
|
join_sql = " AND ".join(
|
|
f'n.{_quote_ident(col)} = o.{_quote_ident(col)}'
|
|
for col in pk_columns
|
|
)
|
|
null_check_col = _quote_ident(pk_columns[0])
|
|
with engine.connect() as conn:
|
|
conn.execute(
|
|
text(
|
|
f"INSERT INTO {_quote_ident(new_name)} ({columns_sql}) "
|
|
f"SELECT {', '.join(f'o.{_quote_ident(col)}' for col in shared_columns)} "
|
|
f"FROM {_quote_ident(old_name)} o "
|
|
f"LEFT JOIN {_quote_ident(new_name)} n ON {join_sql} "
|
|
f"WHERE n.{null_check_col} IS NULL"
|
|
)
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _migrate_legacy_table_rows() -> None:
|
|
for old_name, new_name in LEGACY_TABLE_PAIRS:
|
|
_copy_legacy_table_rows(old_name, new_name)
|
|
|
|
|
|
def _topic_fk_target(table_name: str, constrained_column: str = "bot_id") -> str | None:
|
|
inspector = inspect(engine)
|
|
if not inspector.has_table(table_name):
|
|
return None
|
|
for fk in inspector.get_foreign_keys(table_name):
|
|
cols = [str(col) for col in (fk.get("constrained_columns") or []) if col]
|
|
if cols == [constrained_column]:
|
|
referred = fk.get("referred_table")
|
|
return str(referred) if referred else None
|
|
return None
|
|
|
|
|
|
def _repair_postgres_topic_foreign_keys() -> None:
|
|
if engine.dialect.name != "postgresql":
|
|
return
|
|
targets = {
|
|
"topic_topic": "topic_topic_bot_id_fkey",
|
|
"topic_item": "topic_item_bot_id_fkey",
|
|
}
|
|
with engine.connect() as conn:
|
|
changed = False
|
|
for table_name, constraint_name in targets.items():
|
|
if _topic_fk_target(table_name) == BOT_INSTANCE_TABLE:
|
|
continue
|
|
conn.execute(
|
|
text(
|
|
f'ALTER TABLE {_quote_ident(table_name)} '
|
|
f'DROP CONSTRAINT IF EXISTS {_quote_ident(constraint_name)}'
|
|
)
|
|
)
|
|
conn.execute(
|
|
text(
|
|
f'ALTER TABLE {_quote_ident(table_name)} '
|
|
f'ADD CONSTRAINT {_quote_ident(constraint_name)} '
|
|
f'FOREIGN KEY ({_quote_ident("bot_id")}) '
|
|
f'REFERENCES {_quote_ident(BOT_INSTANCE_TABLE)}({_quote_ident("id")}) '
|
|
f'ON DELETE CASCADE'
|
|
)
|
|
)
|
|
changed = True
|
|
if changed:
|
|
conn.commit()
|
|
|
|
|
|
def _legacy_rows_missing_in_new(old_name: str, new_name: str) -> int:
|
|
inspector = inspect(engine)
|
|
if not inspector.has_table(old_name) or not inspector.has_table(new_name):
|
|
return 0
|
|
pk = inspector.get_pk_constraint(new_name) or {}
|
|
pk_columns = [
|
|
str(col)
|
|
for col in (pk.get("constrained_columns") or [])
|
|
if col
|
|
]
|
|
if not pk_columns:
|
|
return _table_row_count(old_name)
|
|
join_sql = " AND ".join(
|
|
f'n.{_quote_ident(col)} = o.{_quote_ident(col)}'
|
|
for col in pk_columns
|
|
)
|
|
null_check_col = _quote_ident(pk_columns[0])
|
|
with engine.connect() as conn:
|
|
value = conn.execute(
|
|
text(
|
|
f'SELECT COUNT(*) FROM {_quote_ident(old_name)} o '
|
|
f'LEFT JOIN {_quote_ident(new_name)} n ON {join_sql} '
|
|
f'WHERE n.{null_check_col} IS NULL'
|
|
)
|
|
).scalar()
|
|
return int(value or 0)
|
|
|
|
|
|
def _drop_legacy_tables() -> None:
|
|
droppable = [
|
|
old_name
|
|
for old_name, new_name in LEGACY_TABLE_PAIRS
|
|
if _legacy_rows_missing_in_new(old_name, new_name) <= 0
|
|
]
|
|
if not droppable:
|
|
return
|
|
with engine.connect() as conn:
|
|
for old_name in droppable:
|
|
if engine.dialect.name == "postgresql":
|
|
conn.execute(text(f'DROP TABLE IF EXISTS {_quote_ident(old_name)} CASCADE'))
|
|
else:
|
|
conn.execute(text(f'DROP TABLE IF EXISTS {_quote_ident(old_name)}'))
|
|
conn.commit()
|
|
|
|
|
|
def _ensure_botinstance_columns() -> None:
|
|
dialect = engine.dialect.name
|
|
required_columns = {
|
|
"current_state": {
|
|
"sqlite": "TEXT DEFAULT 'IDLE'",
|
|
"postgresql": "TEXT DEFAULT 'IDLE'",
|
|
"mysql": "VARCHAR(64) DEFAULT 'IDLE'",
|
|
},
|
|
"last_action": {
|
|
"sqlite": "TEXT",
|
|
"postgresql": "TEXT",
|
|
"mysql": "LONGTEXT",
|
|
},
|
|
"image_tag": {
|
|
"sqlite": "TEXT DEFAULT 'nanobot-base:v0.1.4'",
|
|
"postgresql": "TEXT DEFAULT 'nanobot-base:v0.1.4'",
|
|
"mysql": "VARCHAR(255) DEFAULT 'nanobot-base:v0.1.4'",
|
|
},
|
|
"access_password": {
|
|
"sqlite": "TEXT DEFAULT ''",
|
|
"postgresql": "TEXT DEFAULT ''",
|
|
"mysql": "VARCHAR(255) DEFAULT ''",
|
|
},
|
|
"enabled": {
|
|
"sqlite": "INTEGER NOT NULL DEFAULT 1",
|
|
"postgresql": "BOOLEAN NOT NULL DEFAULT TRUE",
|
|
"mysql": "BOOLEAN NOT NULL DEFAULT TRUE",
|
|
},
|
|
}
|
|
|
|
inspector = inspect(engine)
|
|
if not inspector.has_table(BOT_INSTANCE_TABLE):
|
|
return
|
|
with engine.connect() as conn:
|
|
existing = {
|
|
str(row.get("name"))
|
|
for row in inspect(conn).get_columns(BOT_INSTANCE_TABLE)
|
|
if row.get("name")
|
|
}
|
|
for col, ddl_map in required_columns.items():
|
|
if col in existing:
|
|
continue
|
|
ddl = ddl_map.get(dialect) or ddl_map.get("sqlite")
|
|
conn.execute(text(f"ALTER TABLE {BOT_INSTANCE_TABLE} ADD COLUMN {col} {ddl}"))
|
|
if "enabled" in existing:
|
|
if dialect == "sqlite":
|
|
conn.execute(text(f"UPDATE {BOT_INSTANCE_TABLE} SET enabled = 1 WHERE enabled IS NULL"))
|
|
else:
|
|
conn.execute(text(f"UPDATE {BOT_INSTANCE_TABLE} SET enabled = TRUE WHERE enabled IS NULL"))
|
|
conn.commit()
|
|
|
|
|
|
def _drop_legacy_botinstance_columns() -> None:
|
|
legacy_columns = [
|
|
"avatar_model",
|
|
"avatar_skin",
|
|
"system_prompt",
|
|
"soul_md",
|
|
"agents_md",
|
|
"user_md",
|
|
"tools_md",
|
|
"tools_config_json",
|
|
"identity_md",
|
|
"llm_provider",
|
|
"llm_model",
|
|
"api_key",
|
|
"api_base",
|
|
"temperature",
|
|
"top_p",
|
|
"max_tokens",
|
|
"presence_penalty",
|
|
"frequency_penalty",
|
|
"send_progress",
|
|
"send_tool_hints",
|
|
"bot_env_json",
|
|
]
|
|
with engine.connect() as conn:
|
|
existing = {
|
|
str(col.get("name"))
|
|
for col in inspect(conn).get_columns(BOT_INSTANCE_TABLE)
|
|
if col.get("name")
|
|
}
|
|
for col in legacy_columns:
|
|
if col not in existing:
|
|
continue
|
|
try:
|
|
if engine.dialect.name == "mysql":
|
|
conn.execute(text(f"ALTER TABLE {BOT_INSTANCE_TABLE} DROP COLUMN `{col}`"))
|
|
elif engine.dialect.name == "sqlite":
|
|
conn.execute(text(f'ALTER TABLE {BOT_INSTANCE_TABLE} DROP COLUMN "{col}"'))
|
|
else:
|
|
conn.execute(text(f'ALTER TABLE {BOT_INSTANCE_TABLE} DROP COLUMN IF EXISTS "{col}"'))
|
|
except Exception:
|
|
# Keep startup resilient on mixed/legacy database engines.
|
|
continue
|
|
conn.commit()
|
|
|
|
|
|
def _ensure_botmessage_columns() -> None:
|
|
if engine.dialect.name != "sqlite":
|
|
return
|
|
required_columns = {
|
|
"media_json": "TEXT",
|
|
"feedback": "TEXT",
|
|
"feedback_at": "DATETIME",
|
|
}
|
|
with engine.connect() as conn:
|
|
existing_rows = conn.execute(text(f"PRAGMA table_info({BOT_MESSAGE_TABLE})")).fetchall()
|
|
existing = {str(row[1]) for row in existing_rows}
|
|
for col, ddl in required_columns.items():
|
|
if col in existing:
|
|
continue
|
|
conn.execute(text(f"ALTER TABLE {BOT_MESSAGE_TABLE} ADD COLUMN {col} {ddl}"))
|
|
conn.commit()
|
|
|
|
|
|
def _drop_legacy_skill_tables() -> None:
|
|
"""Drop deprecated skill registry tables (moved to workspace filesystem mode)."""
|
|
with engine.connect() as conn:
|
|
conn.execute(text("DROP TABLE IF EXISTS botskillmapping"))
|
|
conn.execute(text("DROP TABLE IF EXISTS skillregistry"))
|
|
conn.commit()
|
|
|
|
|
|
def _ensure_sys_setting_columns() -> None:
|
|
dialect = engine.dialect.name
|
|
required_columns = {
|
|
"name": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT ''",
|
|
"postgresql": "TEXT NOT NULL DEFAULT ''",
|
|
"mysql": "VARCHAR(200) NOT NULL DEFAULT ''",
|
|
},
|
|
"category": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT 'general'",
|
|
"postgresql": "TEXT NOT NULL DEFAULT 'general'",
|
|
"mysql": "VARCHAR(64) NOT NULL DEFAULT 'general'",
|
|
},
|
|
"description": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT ''",
|
|
"postgresql": "TEXT NOT NULL DEFAULT ''",
|
|
"mysql": "LONGTEXT",
|
|
},
|
|
"value_type": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT 'json'",
|
|
"postgresql": "TEXT NOT NULL DEFAULT 'json'",
|
|
"mysql": "VARCHAR(32) NOT NULL DEFAULT 'json'",
|
|
},
|
|
"is_public": {
|
|
"sqlite": "INTEGER NOT NULL DEFAULT 0",
|
|
"postgresql": "BOOLEAN NOT NULL DEFAULT FALSE",
|
|
"mysql": "BOOLEAN NOT NULL DEFAULT FALSE",
|
|
},
|
|
"sort_order": {
|
|
"sqlite": "INTEGER NOT NULL DEFAULT 100",
|
|
"postgresql": "INTEGER NOT NULL DEFAULT 100",
|
|
"mysql": "INTEGER NOT NULL DEFAULT 100",
|
|
},
|
|
}
|
|
inspector = inspect(engine)
|
|
if not inspector.has_table(SYS_SETTING_TABLE):
|
|
return
|
|
with engine.connect() as conn:
|
|
existing = {
|
|
str(row.get("name"))
|
|
for row in inspect(conn).get_columns(SYS_SETTING_TABLE)
|
|
if row.get("name")
|
|
}
|
|
for col, ddl_map in required_columns.items():
|
|
if col in existing:
|
|
continue
|
|
ddl = ddl_map.get(dialect) or ddl_map.get("sqlite")
|
|
conn.execute(text(f"ALTER TABLE {SYS_SETTING_TABLE} ADD COLUMN {col} {ddl}"))
|
|
conn.commit()
|
|
|
|
|
|
def _ensure_bot_request_usage_columns() -> None:
|
|
dialect = engine.dialect.name
|
|
required_columns = {
|
|
"message_id": {
|
|
"sqlite": "INTEGER",
|
|
"postgresql": "INTEGER",
|
|
"mysql": "INTEGER",
|
|
},
|
|
"provider": {
|
|
"sqlite": "TEXT",
|
|
"postgresql": "TEXT",
|
|
"mysql": "VARCHAR(120)",
|
|
},
|
|
"model": {
|
|
"sqlite": "TEXT",
|
|
"postgresql": "TEXT",
|
|
"mysql": "VARCHAR(255)",
|
|
},
|
|
}
|
|
inspector = inspect(engine)
|
|
if not inspector.has_table(BOT_REQUEST_USAGE_TABLE):
|
|
return
|
|
with engine.connect() as conn:
|
|
existing = {
|
|
str(row.get("name"))
|
|
for row in inspect(conn).get_columns(BOT_REQUEST_USAGE_TABLE)
|
|
if row.get("name")
|
|
}
|
|
for col, ddl_map in required_columns.items():
|
|
if col in existing:
|
|
continue
|
|
ddl = ddl_map.get(dialect) or ddl_map.get("sqlite")
|
|
conn.execute(text(f"ALTER TABLE {BOT_REQUEST_USAGE_TABLE} ADD COLUMN {col} {ddl}"))
|
|
conn.commit()
|
|
|
|
|
|
def _ensure_topic_tables_sqlite() -> None:
|
|
if engine.dialect.name != "sqlite":
|
|
return
|
|
with engine.connect() as conn:
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS topic_topic (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
bot_id TEXT NOT NULL,
|
|
topic_key TEXT NOT NULL,
|
|
name TEXT NOT NULL DEFAULT '',
|
|
description TEXT NOT NULL DEFAULT '',
|
|
is_active INTEGER NOT NULL DEFAULT 1,
|
|
is_default_fallback INTEGER NOT NULL DEFAULT 0,
|
|
routing_json TEXT NOT NULL DEFAULT '{}',
|
|
view_schema_json TEXT NOT NULL DEFAULT '{}',
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY(bot_id) REFERENCES bot_instance(id)
|
|
)
|
|
"""
|
|
)
|
|
)
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS topic_item (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
bot_id TEXT NOT NULL,
|
|
topic_key TEXT NOT NULL,
|
|
title TEXT NOT NULL DEFAULT '',
|
|
content TEXT NOT NULL DEFAULT '',
|
|
level TEXT NOT NULL DEFAULT 'info',
|
|
tags_json TEXT,
|
|
view_json TEXT,
|
|
source TEXT NOT NULL DEFAULT 'mcp',
|
|
dedupe_key TEXT,
|
|
is_read INTEGER NOT NULL DEFAULT 0,
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY(bot_id) REFERENCES bot_instance(id)
|
|
)
|
|
"""
|
|
)
|
|
)
|
|
|
|
conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS uq_topic_topic_bot_topic_key ON topic_topic(bot_id, topic_key)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_topic_bot_id ON topic_topic(bot_id)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_topic_topic_key ON topic_topic(topic_key)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_topic_bot_fallback ON topic_topic(bot_id, is_default_fallback)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_item_bot_id ON topic_item(bot_id)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_item_topic_key ON topic_item(topic_key)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_item_level ON topic_item(level)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_item_source ON topic_item(source)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_item_is_read ON topic_item(is_read)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_item_created_at ON topic_item(created_at)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_item_bot_topic_created_at ON topic_item(bot_id, topic_key, created_at)"))
|
|
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_topic_item_bot_dedupe ON topic_item(bot_id, dedupe_key)"))
|
|
conn.commit()
|
|
|
|
|
|
def _ensure_topic_columns() -> None:
|
|
dialect = engine.dialect.name
|
|
required_columns = {
|
|
"topic_topic": {
|
|
"name": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT ''",
|
|
"postgresql": "TEXT NOT NULL DEFAULT ''",
|
|
"mysql": "VARCHAR(255) NOT NULL DEFAULT ''",
|
|
},
|
|
"description": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT ''",
|
|
"postgresql": "TEXT NOT NULL DEFAULT ''",
|
|
"mysql": "LONGTEXT",
|
|
},
|
|
"is_active": {
|
|
"sqlite": "INTEGER NOT NULL DEFAULT 1",
|
|
"postgresql": "BOOLEAN NOT NULL DEFAULT TRUE",
|
|
"mysql": "BOOLEAN NOT NULL DEFAULT TRUE",
|
|
},
|
|
"is_default_fallback": {
|
|
"sqlite": "INTEGER NOT NULL DEFAULT 0",
|
|
"postgresql": "BOOLEAN NOT NULL DEFAULT FALSE",
|
|
"mysql": "BOOLEAN NOT NULL DEFAULT FALSE",
|
|
},
|
|
"routing_json": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT '{}'",
|
|
"postgresql": "TEXT NOT NULL DEFAULT '{}'",
|
|
"mysql": "LONGTEXT",
|
|
},
|
|
"view_schema_json": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT '{}'",
|
|
"postgresql": "TEXT NOT NULL DEFAULT '{}'",
|
|
"mysql": "LONGTEXT",
|
|
},
|
|
"created_at": {
|
|
"sqlite": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
|
|
"postgresql": "TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP",
|
|
"mysql": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
|
|
},
|
|
"updated_at": {
|
|
"sqlite": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
|
|
"postgresql": "TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP",
|
|
"mysql": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
|
|
},
|
|
},
|
|
"topic_item": {
|
|
"title": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT ''",
|
|
"postgresql": "TEXT NOT NULL DEFAULT ''",
|
|
"mysql": "VARCHAR(2000) NOT NULL DEFAULT ''",
|
|
},
|
|
"level": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT 'info'",
|
|
"postgresql": "TEXT NOT NULL DEFAULT 'info'",
|
|
"mysql": "VARCHAR(32) NOT NULL DEFAULT 'info'",
|
|
},
|
|
"tags_json": {
|
|
"sqlite": "TEXT",
|
|
"postgresql": "TEXT",
|
|
"mysql": "LONGTEXT",
|
|
},
|
|
"view_json": {
|
|
"sqlite": "TEXT",
|
|
"postgresql": "TEXT",
|
|
"mysql": "LONGTEXT",
|
|
},
|
|
"source": {
|
|
"sqlite": "TEXT NOT NULL DEFAULT 'mcp'",
|
|
"postgresql": "TEXT NOT NULL DEFAULT 'mcp'",
|
|
"mysql": "VARCHAR(64) NOT NULL DEFAULT 'mcp'",
|
|
},
|
|
"dedupe_key": {
|
|
"sqlite": "TEXT",
|
|
"postgresql": "TEXT",
|
|
"mysql": "VARCHAR(200)",
|
|
},
|
|
"is_read": {
|
|
"sqlite": "INTEGER NOT NULL DEFAULT 0",
|
|
"postgresql": "BOOLEAN NOT NULL DEFAULT FALSE",
|
|
"mysql": "BOOLEAN NOT NULL DEFAULT FALSE",
|
|
},
|
|
"created_at": {
|
|
"sqlite": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
|
|
"postgresql": "TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP",
|
|
"mysql": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
|
|
},
|
|
},
|
|
}
|
|
|
|
inspector = inspect(engine)
|
|
with engine.connect() as conn:
|
|
for table_name, cols in required_columns.items():
|
|
if not inspector.has_table(table_name):
|
|
continue
|
|
existing = {
|
|
str(row.get("name"))
|
|
for row in inspector.get_columns(table_name)
|
|
if row.get("name")
|
|
}
|
|
for col, ddl_map in cols.items():
|
|
if col in existing:
|
|
continue
|
|
ddl = ddl_map.get(dialect) or ddl_map.get("sqlite")
|
|
conn.execute(text(f"ALTER TABLE {table_name} ADD COLUMN {col} {ddl}"))
|
|
conn.commit()
|
|
|
|
|
|
def _ensure_topic_indexes() -> None:
|
|
required_indexes = [
|
|
("uq_topic_topic_bot_topic_key", "topic_topic", ["bot_id", "topic_key"], True),
|
|
("idx_topic_topic_bot_id", "topic_topic", ["bot_id"], False),
|
|
("idx_topic_topic_topic_key", "topic_topic", ["topic_key"], False),
|
|
("idx_topic_topic_bot_fallback", "topic_topic", ["bot_id", "is_default_fallback"], False),
|
|
("idx_topic_item_bot_id", "topic_item", ["bot_id"], False),
|
|
("idx_topic_item_topic_key", "topic_item", ["topic_key"], False),
|
|
("idx_topic_item_level", "topic_item", ["level"], False),
|
|
("idx_topic_item_source", "topic_item", ["source"], False),
|
|
("idx_topic_item_is_read", "topic_item", ["is_read"], False),
|
|
("idx_topic_item_created_at", "topic_item", ["created_at"], False),
|
|
("idx_topic_item_bot_topic_created_at", "topic_item", ["bot_id", "topic_key", "created_at"], False),
|
|
("idx_topic_item_bot_dedupe", "topic_item", ["bot_id", "dedupe_key"], False),
|
|
]
|
|
inspector = inspect(engine)
|
|
with engine.connect() as conn:
|
|
for name, table_name, columns, unique in required_indexes:
|
|
if not inspector.has_table(table_name):
|
|
continue
|
|
existing = {
|
|
str(item.get("name"))
|
|
for item in inspector.get_indexes(table_name)
|
|
if item.get("name")
|
|
}
|
|
existing.update(
|
|
str(item.get("name"))
|
|
for item in inspector.get_unique_constraints(table_name)
|
|
if item.get("name")
|
|
)
|
|
if name in existing:
|
|
continue
|
|
unique_sql = "UNIQUE " if unique else ""
|
|
cols_sql = ", ".join(columns)
|
|
conn.execute(text(f"CREATE {unique_sql}INDEX {name} ON {table_name} ({cols_sql})"))
|
|
conn.commit()
|
|
|
|
|
|
def _drop_obsolete_topic_tables() -> None:
|
|
with engine.connect() as conn:
|
|
if engine.dialect.name == "postgresql":
|
|
conn.execute(text('DROP TABLE IF EXISTS "topic_bot_settings"'))
|
|
elif engine.dialect.name == "mysql":
|
|
conn.execute(text("DROP TABLE IF EXISTS `topic_bot_settings`"))
|
|
else:
|
|
conn.execute(text('DROP TABLE IF EXISTS "topic_bot_settings"'))
|
|
conn.commit()
|
|
|
|
|
|
def _cleanup_legacy_default_topics() -> None:
|
|
"""
|
|
Remove legacy auto-created fallback topic rows from early topic-feed design.
|
|
|
|
Historical rows look like:
|
|
- topic_key = inbox
|
|
- name = Inbox
|
|
- description = Default topic for uncategorized items
|
|
- routing_json contains "Fallback topic"
|
|
"""
|
|
with engine.connect() as conn:
|
|
legacy_rows = conn.execute(
|
|
text(
|
|
"""
|
|
SELECT bot_id, topic_key
|
|
FROM topic_topic
|
|
WHERE lower(coalesce(topic_key, '')) = 'inbox'
|
|
AND lower(coalesce(name, '')) = 'inbox'
|
|
AND lower(coalesce(description, '')) = 'default topic for uncategorized items'
|
|
AND lower(coalesce(routing_json, '')) LIKE '%fallback topic%'
|
|
"""
|
|
)
|
|
).fetchall()
|
|
if not legacy_rows:
|
|
return
|
|
for row in legacy_rows:
|
|
bot_id = str(row[0] or "").strip()
|
|
topic_key = str(row[1] or "").strip().lower()
|
|
if not bot_id or not topic_key:
|
|
continue
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
DELETE FROM topic_item
|
|
WHERE bot_id = :bot_id AND lower(coalesce(topic_key, '')) = :topic_key
|
|
"""
|
|
),
|
|
{"bot_id": bot_id, "topic_key": topic_key},
|
|
)
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
DELETE FROM topic_topic
|
|
WHERE bot_id = :bot_id AND lower(coalesce(topic_key, '')) = :topic_key
|
|
"""
|
|
),
|
|
{"bot_id": bot_id, "topic_key": topic_key},
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def align_postgres_sequences() -> None:
|
|
if engine.dialect.name != "postgresql":
|
|
return
|
|
sequence_targets = [
|
|
(BOT_MESSAGE_TABLE, "id"),
|
|
(BOT_REQUEST_USAGE_TABLE, "id"),
|
|
(BOT_ACTIVITY_EVENT_TABLE, "id"),
|
|
("skill_market_item", "id"),
|
|
("bot_skill_install", "id"),
|
|
]
|
|
with engine.connect() as conn:
|
|
for table_name, column_name in sequence_targets:
|
|
seq_name = conn.execute(
|
|
text("SELECT pg_get_serial_sequence(:table_name, :column_name)"),
|
|
{"table_name": table_name, "column_name": column_name},
|
|
).scalar()
|
|
if not seq_name:
|
|
continue
|
|
max_id = conn.execute(
|
|
text(f'SELECT COALESCE(MAX("{column_name}"), 0) FROM "{table_name}"')
|
|
).scalar()
|
|
max_id = int(max_id or 0)
|
|
conn.execute(
|
|
text("SELECT setval(:seq_name, :next_value, :is_called)"),
|
|
{
|
|
"seq_name": seq_name,
|
|
"next_value": max_id if max_id > 0 else 1,
|
|
"is_called": max_id > 0,
|
|
},
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def init_database() -> None:
|
|
lock_conn = _acquire_migration_lock()
|
|
try:
|
|
_rename_legacy_tables()
|
|
SQLModel.metadata.create_all(engine)
|
|
_migrate_legacy_table_rows()
|
|
_drop_legacy_skill_tables()
|
|
_ensure_sys_setting_columns()
|
|
_ensure_bot_request_usage_columns()
|
|
_ensure_botinstance_columns()
|
|
_drop_legacy_botinstance_columns()
|
|
_ensure_botmessage_columns()
|
|
_ensure_topic_tables_sqlite()
|
|
_repair_postgres_topic_foreign_keys()
|
|
_ensure_topic_columns()
|
|
_ensure_topic_indexes()
|
|
_drop_obsolete_topic_tables()
|
|
_cleanup_legacy_default_topics()
|
|
_drop_legacy_tables()
|
|
align_postgres_sequences()
|
|
finally:
|
|
_release_migration_lock(lock_conn)
|
|
|
|
|
|
def get_session():
|
|
with Session(engine) as session:
|
|
yield session
|