Separating database users for schema management and application runtime

[bay-a-dev]

Hello,

We are using Synapse in our company, and for security and traceability reasons, we would like to configure Synapse with two different database users:

• Admin DB user: with full privileges (including schema modification). This user would only be used during installation and during schema upgrade operations when upgrading Synapse.

• Application DB user: with restricted privileges (only data access — no schema modification roles). This user would be used by the running Synapse application in production.

I also tested a possible workaround for this problem by:

• Running an external job with the admin DB user to perform the schema upgrade.

• Configuring Synapse itself to run with the restricted DB user (no schema modification privileges).

However, when I tried this approach, I realized that Synapse still requires a DB user with full privileges (including schema modifications) in order to start properly.

Could you please consider supporting this request of using two different DB users, or suggest an alternative temporary solution to achieve a similar separation?

[anthonyCarigny]

A solution could be to check if the table related to schema version exist BEFORE attempting to run the schema_version.sql file.

def _get_or_create_schema_state(
        txn: Cursor, database_engine: BaseDatabaseEngine
) -> Optional[_SchemaState]:
    # Ensure the schema_version-related tables exist. To avoid requiring CREATE
    # privilege on Postgres when the tables already exist, first check for their
    # presence and only run the creation script if any are missing.
    def _table_exists(name: str) -> bool:
        if isinstance(database_engine, PostgresEngine):
            # Use to_regclass which returns NULL if the relation does not exist
            txn.execute("SELECT to_regclass(?) IS NOT NULL", (name,))
            row = txn.fetchone()
            return bool(row and row[0])

    required_tables = (
        "schema_version",
        "schema_compat_version",
        "applied_schema_deltas",
        "applied_module_schemas",
    )

    missing = [t for t in required_tables if not _table_exists(t)]

    if missing:
        sql_path = os.path.join(schema_path, "common", "schema_version.sql")
        database_engine.execute_script_file(txn, sql_path)

    txn.execute("SELECT version, upgraded FROM schema_version")
    row = txn.fetchone()

    if row is None:
        # new database
        return None

    current_version = int(row[0])
    upgraded = bool(row[1])

    compat_version: Optional[int] = None
    txn.execute("SELECT compat_version FROM schema_compat_version")
    row = txn.fetchone()
    if row is not None:
        compat_version = int(row[0])

    txn.execute(
        "SELECT file FROM applied_schema_deltas WHERE version >= ?",
        (current_version,),
    )
    applied_deltas = tuple(d for (d,) in txn)

    return _SchemaState(
        current_version=current_version,
        compat_version=compat_version,
        applied_deltas=applied_deltas,
        upgraded=upgraded,
    )