im_server/server/db/postgres/adapter.go

//go:build postgres
// +build postgres

// Package postgres is a database adapter for PostgreSQL.
package postgres

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"hash/fnv"
	"log"
	"net/url"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/jmoiron/sqlx"
	"github.com/tinode/chat/server/auth"
	"github.com/tinode/chat/server/db/common"
	"github.com/tinode/chat/server/store"
	t "github.com/tinode/chat/server/store/types"
)

// adapter holds MySQL connection data.
type adapter struct {
	db         *pgxpool.Pool
	poolConfig *pgxpool.Config
	dsn        string
	dbName     string
	// Maximum number of records to return
	maxResults int
	// Maximum number of message records to return
	maxMessageResults int
	version           int

	// Single query timeout.
	sqlTimeout time.Duration
	// DB transaction timeout.
	txTimeout time.Duration
}

const (
	adpVersion  = 116
	adapterName = "postgres"

	defaultMaxResults = 1024
	// This is capped by the Session's send queue limit (128).
	defaultMaxMessageResults = 100

	// If DB request timeout is specified,
	// we allocate txTimeoutMultiplier times more time for transactions.
	txTimeoutMultiplier = 1.5
)

type configType struct {
	// DB connection settings:
	// Using fields
	User   string `json:"user,omitempty"`
	Passwd string `json:"passwd,omitempty"`
	Host   string `json:"host,omitempty"`
	Port   string `json:"port,omitempty"`
	DBName string `json:"dbname,omitempty"`
	// Deprecated.
	DSN string `json:"dsn,omitempty"`

	// Connection pool settings.
	//
	// Maximum number of open connections to the database.
	MaxOpenConns int `json:"max_open_conns,omitempty"`
	// Maximum number of connections in the idle connection pool.
	MaxIdleConns int `json:"max_idle_conns,omitempty"`
	// Maximum amount of time a connection may be reused (in seconds).
	ConnMaxLifetime int `json:"conn_max_lifetime,omitempty"`

	// SSL mode determines how SSL connections are handled.
	// Supported values:
	//   - "disable": No SSL connection (default)
	//   - "require": Require SSL connection but don't verify server certificate
	//   - "verify-ca": Require SSL and verify that the server certificate is issued by a trusted CA
	//   - "verify-full": Require SSL and verify that the server certificate matches the server hostname
	//   - "prefer": Try SSL first, fallback to non-SSL if SSL fails
	//   - "allow": Try non-SSL first, fallback to SSL if non-SSL fails
	SSLMode string `json:"ssl_mode,omitempty"`

	// DB request timeout (in seconds).
	// If 0 (or negative), no timeout is applied.
	SqlTimeout int `json:"sql_timeout,omitempty"`
}

func (a *adapter) getContext() (context.Context, context.CancelFunc) {
	if a.sqlTimeout > 0 {
		return context.WithTimeout(context.Background(), a.sqlTimeout)
	}
	return context.Background(), nil
}

func (a *adapter) getContextForTx() (context.Context, context.CancelFunc) {
	if a.txTimeout > 0 {
		return context.WithTimeout(context.Background(), a.txTimeout)
	}
	return context.Background(), nil
}

// Open initializes database session
func (a *adapter) Open(jsonconfig json.RawMessage) error {
	if a.db != nil {
		return errors.New("postgres adapter is already connected")
	}

	if len(jsonconfig) < 2 {
		return errors.New("postgres adapter missing config")
	}

	var err error
	var config configType
	ctx := context.Background()
	if err = json.Unmarshal(jsonconfig, &config); err != nil {
		return errors.New("postgres adapter failed to parse config: " + err.Error())
	}

	if config.DSN != "" {
		a.dsn = config.DSN
		if uri, err := url.Parse(a.dsn); err == nil {
			a.dbName = strings.TrimPrefix(uri.Path, "/")
		} else {
			return err
		}
	} else {
		if a.dsn, err = setConnStr(config); err != nil {
			return err
		}
		a.dbName = config.DBName
	}

	if a.maxResults <= 0 {
		a.maxResults = defaultMaxResults
	}

	if a.maxMessageResults <= 0 {
		a.maxMessageResults = defaultMaxMessageResults
	}

	if a.poolConfig, err = pgxpool.ParseConfig(a.dsn); err != nil {
		return errors.New("postgres adapter failed to parse DSN: " + err.Error())
	}

	// NewWithConfig only creates a pool object. Ping to validate connectivity.
	a.db, err = pgxpool.NewWithConfig(ctx, a.poolConfig)
	if err != nil {
		return err
	}

	err = a.db.Ping(ctx)
	if isMissingDb(err) {
		// Missing DB is OK if we are initializing the database.
		// Since tinode DB does not exist, connect without specifying the DB name.
		a.db.Close()
		a.poolConfig.ConnConfig.Database = ""
		a.db, err = pgxpool.NewWithConfig(ctx, a.poolConfig)
		if err != nil {
			return err
		}
		err = a.db.Ping(ctx)
	}
	if err != nil {
		return err
	}

	if err == nil {
		if config.MaxOpenConns > 0 {
			a.poolConfig.MaxConns = int32(config.MaxOpenConns)
		}
		if config.MaxIdleConns > 0 {
			a.poolConfig.MinConns = int32(config.MaxIdleConns)
		}
		if config.ConnMaxLifetime > 0 {
			a.poolConfig.MaxConnLifetime = time.Duration(config.ConnMaxLifetime) * time.Second
		}
		if config.SqlTimeout > 0 {
			a.sqlTimeout = time.Duration(config.SqlTimeout) * time.Second
			// We allocate txTimeoutMultiplier times sqlTimeout for transactions.
			a.txTimeout = time.Duration(float64(config.SqlTimeout)*txTimeoutMultiplier) * time.Second
		}
	}
	return err
}

// Close closes the underlying database connection
func (a *adapter) Close() error {
	if a.db != nil {
		a.db.Close()
		a.db = nil
		a.version = -1
	}
	return nil
}

// IsOpen returns true if connection to database has been established. It does not check if
// connection is actually live.
func (a *adapter) IsOpen() bool {
	return a.db != nil
}

// GetDbVersion returns current database version.
func (a *adapter) GetDbVersion() (int, error) {
	if a.version > 0 {
		return a.version, nil
	}

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	var vers string
	err := a.db.QueryRow(ctx, "SELECT value FROM im_kvmeta WHERE key='version'").Scan(&vers)
	if err != nil {
		if isMissingDb(err) || isMissingTable(err) || err == pgx.ErrNoRows {
			err = errors.New("Database not initialized")
		}
		return -1, err
	}

	a.version, _ = strconv.Atoi(vers)

	return a.version, nil
}

func (a *adapter) updateDbVersion(v int) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	a.version = -1
	if _, err := a.db.Exec(ctx, `UPDATE im_kvmeta SET "value"=$1 WHERE "key"='version'`, strconv.Itoa(v)); err != nil {
		return err
	}
	return nil
}

// CheckDbVersion checks whether the actual DB version matches the expected version of this adapter.
func (a *adapter) CheckDbVersion() error {
	version, err := a.GetDbVersion()
	if err != nil {
		return err
	}

	if version != adpVersion {
		return errors.New("Invalid database version " + strconv.Itoa(version) +
			". Expected " + strconv.Itoa(adpVersion))
	}

	return nil
}

// Version returns adapter version.
func (adapter) Version() int {
	return adpVersion
}

// DB connection stats object.
func (a *adapter) Stats() any {
	if a.db == nil {
		return nil
	}
	return a.db.Stat()
}

// GetName returns string that adapter uses to register itself with store.
func (a *adapter) GetName() string {
	return adapterName
}

// SetMaxResults configures how many results can be returned in a single DB call.
func (a *adapter) SetMaxResults(val int) error {
	if val <= 0 {
		a.maxResults = defaultMaxResults
	} else {
		a.maxResults = val
	}

	return nil
}

// CreateDb initializes the storage.
func (a *adapter) CreateDb(reset bool) error {
	var err error
	var tx pgx.Tx

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	// Can't use an existing connection because it's configured with a database name which may not exist.
	// Don't care if it does not close cleanly.
	if a.db != nil {
		a.db.Close()
	}

	// Create default database name
	a.poolConfig.ConnConfig.Database = "postgres"

	a.db, err = pgxpool.NewWithConfig(ctx, a.poolConfig)
	if err != nil {
		return err
	}
	if err = a.db.Ping(ctx); err != nil {
		return err
	}

	if reset {
		if _, err = a.db.Exec(ctx, fmt.Sprintf("DROP DATABASE IF EXISTS %s;", a.dbName)); err != nil {
			return err
		}
	}

	if _, err = a.db.Exec(ctx, fmt.Sprintf("CREATE DATABASE %s WITH ENCODING utf8;", a.dbName)); err != nil {
		return err
	}

	a.poolConfig.ConnConfig.Database = a.dbName
	a.db, err = pgxpool.NewWithConfig(ctx, a.poolConfig)
	if err != nil {
		return err
	}
	if err = a.db.Ping(ctx); err != nil {
		return err
	}

	if tx, err = a.db.Begin(ctx); err != nil {
		return err
	}

	defer func() {
		if err != nil {
			// FIXME: This is useless: MySQL auto-commits on every CREATE TABLE.
			// Maybe DROP DATABASE instead.
			tx.Rollback(ctx)
		}
	}()

	// Indexed im_users.
	if _, err := tx.Exec(ctx,
		`CREATE TABLE im_users(
			id        BIGINT NOT NULL,
			createdat TIMESTAMP(3) NOT NULL,
			updatedat TIMESTAMP(3) NOT NULL,
			state     SMALLINT NOT NULL DEFAULT 0,
			stateat   TIMESTAMP(3),
			access    JSON,
			lastseen  TIMESTAMP,
			useragent VARCHAR(255) DEFAULT '',
			public    JSON,
			trusted   JSON,
			tags      JSON,
			PRIMARY KEY(id)
		);
		CREATE INDEX users_state_stateat ON im_users(state, stateat);
		CREATE INDEX users_lastseen_updatedat ON im_users(lastseen, updatedat);`); err != nil {
		return err
	}

	// Indexed user tags.
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_usertags(
			id     SERIAL NOT NULL,
			userid BIGINT NOT NULL,
			tag    VARCHAR(96) NOT NULL,
			PRIMARY KEY(id),
			FOREIGN KEY(userid) REFERENCES im_users(id)
		);
		CREATE INDEX usertags_tag ON im_usertags(tag);
		CREATE UNIQUE INDEX usertags_userid_tag ON im_usertags(userid, tag);`); err != nil {
		return err
	}

	// Indexed im_devices. Normalized into a separate table.
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_devices(
			id       SERIAL NOT NULL,
			userid   BIGINT NOT NULL,
			hash     CHAR(16) NOT NULL,
			deviceid TEXT NOT NULL,
			platform VARCHAR(32),
			lastseen TIMESTAMP NOT NULL,
			lang     VARCHAR(8),
			PRIMARY KEY(id),
			FOREIGN KEY(userid) REFERENCES im_users(id)
		);
		CREATE UNIQUE INDEX devices_hash ON im_devices(hash);`); err != nil {
		return err
	}

	// Authentication records for the basic authentication scheme.
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_auth(
			id      SERIAL NOT NULL,
			uname   VARCHAR(32) NOT NULL,
			userid  BIGINT NOT NULL,
			scheme  VARCHAR(16) NOT NULL,
			authlvl INT NOT NULL,
			secret  VARCHAR(255) NOT NULL,
			expires TIMESTAMP,
			PRIMARY KEY(id),
			FOREIGN KEY(userid) REFERENCES im_users(id)
		);
		CREATE UNIQUE INDEX auth_userid_scheme ON im_auth(userid, scheme);
		CREATE UNIQUE INDEX auth_uname ON im_auth(uname);`); err != nil {
		return err
	}

	// Topics
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_topics(
			id        SERIAL NOT NULL,
			createdat TIMESTAMP(3) NOT NULL,
			updatedat TIMESTAMP(3) NOT NULL,
			state     SMALLINT NOT NULL DEFAULT 0,
			stateat   TIMESTAMP(3),
			touchedat TIMESTAMP(3),
			name      VARCHAR(25) NOT NULL,
			usebt     BOOLEAN DEFAULT FALSE,
			owner     BIGINT NOT NULL DEFAULT 0,
			access    JSON,
			seqid     INT NOT NULL DEFAULT 0,
			delid     INT DEFAULT 0,
			subcnt    INT DEFAULT 0,
			public    JSON,
			trusted   JSON,
			tags      JSON,
			aux				JSON,
			PRIMARY KEY(id)
		);
		CREATE UNIQUE INDEX topics_name ON im_topics(name);
		CREATE INDEX topics_owner ON im_topics(owner);
		CREATE INDEX topics_state_stateat ON im_topics(state, stateat);
		CREATE INDEX topics_name_state_seqid ON im_topics(name, state, seqid);`); err != nil {
		return err
	}

	// Create system topic 'sys'.
	if err = createSystemTopic(tx); err != nil {
		return err
	}

	// Indexed topic tags.
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_topictags(
			id    SERIAL NOT NULL,
			topic VARCHAR(25) NOT NULL,
			tag   VARCHAR(96) NOT NULL,
			PRIMARY KEY(id),
			FOREIGN KEY(topic) REFERENCES im_topics(name)
		);
		CREATE INDEX topictags_tag ON im_topictags(tag);
		CREATE UNIQUE INDEX topictags_topic_tag ON im_topictags(topic, tag);`); err != nil {
		return err
	}

	// Subscriptions
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_subscriptions(
			id        SERIAL NOT NULL,
			createdat TIMESTAMP(3) NOT NULL,
			updatedat TIMESTAMP(3) NOT NULL,
			deletedat TIMESTAMP(3),
			userid    BIGINT NOT NULL,
			topic     VARCHAR(25) NOT NULL,
			delid     INT DEFAULT 0,
			recvseqid INT DEFAULT 0,
			readseqid INT DEFAULT 0,
			modewant  VARCHAR(8),
			modegiven VARCHAR(8),
			private   JSON,
			PRIMARY KEY(id),
			FOREIGN KEY(userid) REFERENCES im_users(id)
		);
		CREATE UNIQUE INDEX subscriptions_topic_userid ON im_subscriptions(topic, userid);
		CREATE INDEX subscriptions_topic ON im_subscriptions(topic);
		CREATE INDEX subscriptions_deletedat ON im_subscriptions(deletedat);
		CREATE INDEX subscriptions_userid_topic_deletedat ON im_subscriptions(userid, topic, deletedat);`); err != nil {
		return err
	}

	// Messages
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_messages(
			id        SERIAL NOT NULL,
			createdat TIMESTAMP(3) NOT NULL,
			updatedat TIMESTAMP(3) NOT NULL,
			deletedat TIMESTAMP(3),
			delid     INT DEFAULT 0,
			seqid     INT NOT NULL,
			topic     VARCHAR(25) NOT NULL,
			"from"    BIGINT NOT NULL,
			head      JSON,
			content   JSON,
			PRIMARY KEY(id),
			FOREIGN KEY(topic) REFERENCES im_topics(name)
		);
		CREATE UNIQUE INDEX messages_topic_seqid ON im_messages(topic, seqid);`); err != nil {
		return err
	}

	// Deletion log
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_dellog(
			id         SERIAL NOT NULL,
			topic      VARCHAR(25) NOT NULL,
			deletedfor BIGINT NOT NULL DEFAULT 0,
			delid      INT NOT NULL,
			low        INT NOT NULL,
			hi         INT NOT NULL,
			PRIMARY KEY(id),
			FOREIGN KEY(topic) REFERENCES im_topics(name)
		);
		CREATE INDEX dellog_topic_delid_deletedfor ON im_dellog(topic,delid,deletedfor);
		CREATE INDEX dellog_topic_deletedfor_low_hi ON im_dellog(topic,deletedfor,low,hi);
		CREATE INDEX dellog_deletedfor ON im_dellog(deletedfor);`); err != nil {
		return err
	}

	// User im_credentials
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_credentials(
			id        SERIAL NOT NULL,
			createdat TIMESTAMP(3) NOT NULL,
			updatedat TIMESTAMP(3) NOT NULL,
			deletedat TIMESTAMP(3),
			method    VARCHAR(16) NOT NULL,
			value     VARCHAR(128) NOT NULL,
			synthetic VARCHAR(192) NOT NULL,
			userid    BIGINT NOT NULL,
			resp      VARCHAR(255),
			done      BOOLEAN NOT NULL DEFAULT FALSE,
			retries   INT NOT NULL DEFAULT 0,
			PRIMARY KEY(id),
			FOREIGN KEY(userid) REFERENCES im_users(id)
		);
		CREATE UNIQUE INDEX credentials_uniqueness ON im_credentials(synthetic);`); err != nil {
		return err
	}

	// Records of uploaded files.
	// Don't add FOREIGN KEY on userid. It's not needed and it will break user deletion.
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_fileuploads(
			id        BIGINT NOT NULL,
			createdat TIMESTAMP(3) NOT NULL,
			updatedat TIMESTAMP(3) NOT NULL,
			userid    BIGINT,
			status    INT NOT NULL,
			mimetype  VARCHAR(255) NOT NULL,
			size      BIGINT NOT NULL,
			etag      VARCHAR(128),
			location  VARCHAR(2048) NOT NULL,
			PRIMARY KEY(id)
		);
		CREATE INDEX fileuploads_status ON im_fileuploads(status);`); err != nil {
		return err
	}

	// Links between uploaded files and the im_topics, im_users or im_messages they are attached to.
	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_filemsglinks(
			id        SERIAL NOT NULL,
			createdat TIMESTAMP(3) NOT NULL,
			fileid    BIGINT NOT NULL,
			msgid     INT,
			topic     VARCHAR(25),
			userid    BIGINT,
			PRIMARY KEY(id),
			FOREIGN KEY(fileid) REFERENCES im_fileuploads(id) ON DELETE CASCADE,
			FOREIGN KEY(msgid) REFERENCES im_messages(id) ON DELETE CASCADE,
			FOREIGN KEY(topic) REFERENCES im_topics(name) ON DELETE CASCADE,
			FOREIGN KEY(userid) REFERENCES im_users(id) ON DELETE CASCADE
		);`); err != nil {
		return err
	}

	if _, err = tx.Exec(ctx,
		`CREATE TABLE im_kvmeta(
			"key"     VARCHAR(64) NOT NULL,
			createdat TIMESTAMP(3),
			"value"   TEXT,
			PRIMARY KEY("key")
		);
		CREATE INDEX kvmeta_createdat_key ON im_kvmeta(createdat, "key");`); err != nil {
		return err
	}
	if _, err = tx.Exec(ctx, `INSERT INTO im_kvmeta("key", "value") VALUES($1, $2)`, "version", strconv.Itoa(adpVersion)); err != nil {
		return err
	}

	return tx.Commit(ctx)
}

// UpgradeDb upgrades the database, if necessary.
func (a *adapter) UpgradeDb() error {
	bumpVersion := func(a *adapter, x int) error {
		if err := a.updateDbVersion(x); err != nil {
			return err
		}
		_, err := a.GetDbVersion()
		return err
	}

	if _, err := a.GetDbVersion(); err != nil {
		return err
	}

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	if a.version == 112 {
		// Perform database upgrade from version 112 to version 113.

		// Index for deleting unvalidated accounts.
		if _, err := a.db.Exec(ctx, "CREATE INDEX users_lastseen_updatedat ON im_users(lastseen,updatedat)"); err != nil {
			return err
		}

		// Allow lnger im_kvmeta keys.
		if _, err := a.db.Exec(ctx, `ALTER TABLE im_kvmeta ALTER COLUMN "key" TYPE VARCHAR(64)`); err != nil {
			return err
		}

		if _, err := a.db.Exec(ctx, `ALTER TABLE im_kvmeta ALTER COLUMN "key" SET NOT NULL`); err != nil {
			return err
		}

		// Add timestamp to im_kvmeta.
		if _, err := a.db.Exec(ctx, `ALTER TABLE im_kvmeta ADD COLUMN createdat TIMESTAMP(3)`); err != nil {
			return err
		}

		// Add compound index on the new field and key (could be searched by key prefix).
		if _, err := a.db.Exec(ctx, `CREATE INDEX kvmeta_createdat_key ON im_kvmeta(createdat, "key")`); err != nil {
			return err
		}

		if err := bumpVersion(a, 113); err != nil {
			return err
		}
	}

	if a.version == 113 {
		// Perform database upgrade from version 113 to version 114.

		if _, err := a.db.Exec(ctx, "ALTER TABLE im_topics ADD COLUMN aux JSON"); err != nil {
			return err
		}

		if _, err := a.db.Exec(ctx, "ALTER TABLE im_fileuploads ADD COLUMN etag VARCHAR(128)"); err != nil {
			return err
		}

		if err := bumpVersion(a, 114); err != nil {
			return err
		}
	}

	if a.version == 114 {
		// Perform database upgrade from version 114 to version 115.

		// Find relevant im_subscriptions for given im_users efficiently, and use the join key too.
		if _, err := a.db.Exec(ctx, "CREATE INDEX idx_subs_user_topic_del ON im_subscriptions(userid, topic, deletedat)"); err != nil {
			return err
		}

		// Optimizes join; state filters; seqid supports the SUM operation.
		if _, err := a.db.Exec(ctx, "CREATE INDEX idx_topics_name_state_seqid ON im_topics(name, state, seqid)"); err != nil {
			return err
		}

		if err := bumpVersion(a, 115); err != nil {
			return err
		}
	}

	if a.version == 115 {
		// Perform database upgrade from version 115 to version 116.

		// Add subscriber count column to the im_topics table.
		if _, err := a.db.Exec(ctx, "ALTER TABLE im_topics ADD subcnt INT DEFAULT 0"); err != nil {
			return err
		}

		if err := bumpVersion(a, 116); err != nil {
			return err
		}
	}

	if a.version != adpVersion {
		return errors.New("Failed to perform database upgrade to version " + strconv.Itoa(adpVersion) +
			". DB is still at " + strconv.Itoa(a.version))
	}
	return nil
}

func createSystemTopic(tx pgx.Tx) error {
	now := t.TimeNow()
	query := `INSERT INTO im_topics(createdat,updatedat,state,touchedat,name,access,public)
				VALUES($1,$2,$3,$4,'sys','{"Auth": "N","Anon": "N"}','{"fn": "System"}')`
	_, err := tx.Exec(context.Background(), query, now, now, t.StateOK, now)
	return err
}

func addTags(ctx context.Context, tx pgx.Tx, table, keyName string, keyVal any, tags []string, ignoreDups bool) error {
	if len(tags) == 0 {
		return nil
	}

	//addTags(ctx, tx, "im_usertags", "userid", decoded_uid, add, reset == nil)
	sql := "INSERT INTO " + table + " (" + keyName + ",tag) VALUES($1,$2)"
	if ignoreDups {
		sql += " ON CONFLICT DO NOTHING"
	}
	for _, tag := range tags {
		if _, err := tx.Exec(ctx, sql, keyVal, tag); err != nil {
			if isDupe(err) {
				return t.ErrDuplicate
			}
			return err
		}
	}

	return nil
}

func removeTags(ctx context.Context, tx pgx.Tx, table, keyName string, keyVal any, tags []string) error {
	if len(tags) == 0 {
		return nil
	}

	sql, args := expandQuery("DELETE FROM "+table+" WHERE "+keyName+"=? AND tag IN (?)", keyVal, tags)
	_, err := tx.Exec(ctx, sql, args...)

	return err
}

// UserCreate creates a new user. Returns error and true if error is due to duplicate user name,
// false for any other error
func (a *adapter) UserCreate(user *t.User) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}

	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	decoded_uid := store.DecodeUid(user.Uid())
	if _, err = tx.Exec(ctx,
		"INSERT INTO im_users(id,createdat,updatedat,state,access,public,trusted,tags) VALUES($1,$2,$3,$4,$5,$6,$7,$8);",
		decoded_uid,
		user.CreatedAt,
		user.UpdatedAt,
		user.State,
		user.Access,
		common.ToJSON(user.Public),
		common.ToJSON(user.Trusted),
		user.Tags); err != nil {
		return err
	}

	// Save user's tags to a separate table to make user findable.
	if err = addTags(ctx, tx, "im_usertags", "userid", decoded_uid, user.Tags, false); err != nil {
		return err
	}

	return tx.Commit(ctx)
}

// Add user's authentication record
func (a *adapter) AuthAddRecord(uid t.Uid, scheme, unique string, authLvl auth.Level,
	secret []byte, expires time.Time) error {

	var exp *time.Time
	if !expires.IsZero() {
		exp = &expires
	}
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	if _, err := a.db.Exec(ctx, "INSERT INTO im_auth(uname,userid,scheme,authLvl,secret,expires) VALUES($1,$2,$3,$4,$5,$6)",
		unique, store.DecodeUid(uid), scheme, authLvl, secret, exp); err != nil {
		if isDupe(err) {
			return t.ErrDuplicate
		}
		return err
	}
	return nil
}

// AuthDelScheme deletes an existing authentication scheme for the user.
func (a *adapter) AuthDelScheme(user t.Uid, scheme string) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	_, err := a.db.Exec(ctx, "DELETE FROM im_auth WHERE userid=$1 AND scheme=$2", store.DecodeUid(user), scheme)
	return err
}

// AuthDelAllRecords deletes all authentication records for the user.
func (a *adapter) AuthDelAllRecords(user t.Uid) (int, error) {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	res, err := a.db.Exec(ctx, "DELETE FROM im_auth WHERE userid=$1", store.DecodeUid(user))
	if err != nil {
		return 0, err
	}
	count := res.RowsAffected()

	return int(count), nil
}

// Update user's authentication unique, secret, im_auth level.
func (a *adapter) AuthUpdRecord(uid t.Uid, scheme, unique string, authLvl auth.Level,
	secret []byte, expires time.Time) error {

	parapg := []string{"authLvl=?"}
	args := []any{authLvl}
	if unique != "" {
		parapg = append(parapg, "uname=?")
		args = append(args, unique)
	}
	if len(secret) > 0 {
		parapg = append(parapg, "secret=?")
		args = append(args, secret)
	}
	if !expires.IsZero() {
		parapg = append(parapg, "expires=?")
		args = append(args, expires)
	}
	args = append(args, store.DecodeUid(uid), scheme)

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	sql, args := expandQuery("UPDATE im_auth SET "+strings.Join(parapg, ",")+" WHERE userid=? AND scheme=?", args...)
	resp, err := a.db.Exec(ctx, sql, args...)
	if isDupe(err) {
		return t.ErrDuplicate
	}

	if count := resp.RowsAffected(); count <= 0 {
		return t.ErrNotFound
	}

	return err
}

// Retrieve user's authentication record
func (a *adapter) AuthGetRecord(uid t.Uid, scheme string) (string, auth.Level, []byte, time.Time, error) {
	var expires time.Time

	var record struct {
		Uname   string
		Authlvl auth.Level
		Secret  []byte
		Expires *time.Time
	}

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	if err := a.db.QueryRow(ctx, "SELECT uname,secret,expires,authlvl FROM im_auth WHERE userid=$1 AND scheme=$2",
		store.DecodeUid(uid), scheme).Scan(
		&record.Uname, &record.Secret, &record.Expires, &record.Authlvl); err != nil {
		if err == pgx.ErrNoRows {
			// Nothing found - use standard error.
			err = t.ErrNotFound
		}
		return "", 0, nil, expires, err
	}

	if record.Expires != nil {
		expires = *record.Expires
	}

	return record.Uname, record.Authlvl, record.Secret, expires, nil
}

// Retrieve user's authentication record
func (a *adapter) AuthGetUniqueRecord(unique string) (t.Uid, auth.Level, []byte, time.Time, error) {
	var expires time.Time

	var record struct {
		Userid  int64
		Authlvl auth.Level
		Secret  []byte
		Expires *time.Time
	}

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	if err := a.db.QueryRow(ctx, "SELECT userid,secret,expires,authlvl FROM im_auth WHERE uname=$1", unique).Scan(
		&record.Userid, &record.Secret, &record.Expires, &record.Authlvl); err != nil {
		if err == pgx.ErrNoRows {
			// Nothing found - clear the error
			err = nil
		}
		return t.ZeroUid, 0, nil, expires, err
	}

	if record.Expires != nil {
		expires = *record.Expires
	}

	return store.EncodeUid(record.Userid), record.Authlvl, record.Secret, expires, nil
}

// UserGet fetches a single user by user id. If user is not found it returns (nil, nil)
func (a *adapter) UserGet(uid t.Uid) (*t.User, error) {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	var user t.User
	var id int64
	row, err := a.db.Query(ctx, "SELECT * FROM im_users WHERE id=$1 AND state!=$2", store.DecodeUid(uid), t.StateDeleted)
	if err != nil {
		return nil, err
	}
	defer row.Close()

	if !row.Next() {
		// Nothing found: user does not exist or marked as soft-deleted
		return nil, nil
	}

	err = row.Scan(&id, &user.CreatedAt, &user.UpdatedAt, &user.State, &user.StateAt, &user.Access, &user.LastSeen, &user.UserAgent, &user.Public, &user.Trusted, &user.Tags)
	if err == nil {
		user.SetUid(uid)
		return &user, nil
	}

	return nil, err
}

func (a *adapter) UserGetAll(ids ...t.Uid) ([]t.User, error) {
	uids := make([]any, len(ids))
	for i, id := range ids {
		uids[i] = store.DecodeUid(id)
	}

	im_users := []t.User{}
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	rows, err := a.db.Query(ctx, "SELECT * FROM im_users WHERE id = ANY ($1) AND state!=$2", uids, t.StateDeleted)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	for rows.Next() {
		var user t.User
		var id int64
		if err = rows.Scan(&id, &user.CreatedAt, &user.UpdatedAt, &user.State, &user.StateAt, &user.Access, &user.LastSeen, &user.UserAgent, &user.Public, &user.Trusted, &user.Tags); err != nil {
			im_users = nil
			break
		}
		user.SetUid(store.EncodeUid(id))

		im_users = append(im_users, user)
	}
	if err == nil {
		err = rows.Err()
	}

	return im_users, err
}

// UserDelete deletes specified user: wipes completely (hard-delete) or marks as deleted.
// TODO: report when the user is not found.
func (a *adapter) UserDelete(uid t.Uid, hard bool) error {
	query := "SELECT name FROM im_topics WHERE owner=$1"
	args := []any{store.DecodeUid(uid)}
	// In case of hard delete, delete all im_topics, even those which were
	// soft-deleted previsously.
	if !hard {
		query += " AND state!=$2"
		args = append(args, t.StateDeleted)
	}
	// Get a list of topic names owned by the user (as 'grp' and 'chn').
	ownTopics, err := a.topicNamesForUser(query, false, args...)
	if err != nil {
		return err
	}

	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}

	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	now := t.TimeNow()
	decoded_uid := store.DecodeUid(uid)

	if hard {
		// Delete user's im_devices
		// t.ErrNotFound = user has no im_devices.
		if err = deviceDelete(ctx, tx, uid, ""); err != nil && err != t.ErrNotFound {
			return err
		}

		// Delete user's im_subscriptions in all im_topics.
		if err = subsDelForUser(ctx, tx, decoded_uid, true); err != nil {
			return err
		}

		// Delete records of im_messages soft-deleted for the user.
		if _, err = tx.Exec(ctx, "DELETE FROM im_dellog WHERE deletedfor=$1", decoded_uid); err != nil {
			return err
		}

		// Can't delete user's im_messages in all im_topics because we cannot notify im_topics of such deletion.
		// Just leave the im_messages there marked as sent by "not found" user.

		// Delete im_topics where the user is the owner.

		if len(ownTopics) > 0 {
			// First delete all im_messages in those im_topics.
			if _, err = tx.Exec(ctx, "DELETE FROM im_dellog USING im_topics WHERE im_topics.name=im_dellog.topic AND im_topics.owner=$1",
				decoded_uid); err != nil {
				return err
			}

			// Deletion of im_messages will cascade to im_filemsglinks and so to im_fileuploads.
			if _, err = tx.Exec(ctx, "DELETE FROM im_messages USING im_topics WHERE im_topics.name=im_messages.topic AND im_topics.owner=$1",
				decoded_uid); err != nil {
				return err
			}
			// Delete im_subscriptions for all im_users where the user is the owner of the topic.
			sql, args, _ := sqlx.In("DELETE FROM im_subscriptions AS s WHERE topic IN (?)", ownTopics)
			if _, err = tx.Exec(ctx, sqlx.Rebind(sqlx.DOLLAR, sql), args...); err != nil {
				return err
			}

			// Delete topic tags.
			if _, err = tx.Exec(ctx, "DELETE FROM im_topictags USING im_topics WHERE im_topics.name=im_topictags.topic AND im_topics.owner=$1",
				decoded_uid); err != nil {
				return err
			}

			// And finally delete the im_topics.
			if _, err = tx.Exec(ctx, "DELETE FROM im_topics WHERE owner=$1", decoded_uid); err != nil {
				return err
			}
		}

		// Delete user's authentication records.
		if _, err = tx.Exec(ctx, "DELETE FROM im_auth WHERE userid=$1", decoded_uid); err != nil {
			return err
		}

		// Delete all im_credentials.
		if err = credDel(ctx, tx, uid, "", ""); err != nil && err != t.ErrNotFound {
			return err
		}

		if _, err = tx.Exec(ctx, "DELETE FROM im_usertags WHERE userid=$1", decoded_uid); err != nil {
			return err
		}

		if _, err = tx.Exec(ctx, "DELETE FROM im_users WHERE id=$1", decoded_uid); err != nil {
			return err
		}
	} else {
		// Disable all user's im_subscriptions. That includes p2p im_subscriptions. No need to delete them.
		if err = subsDelForUser(ctx, tx, decoded_uid, false); err != nil {
			return err
		}

		if len(ownTopics) > 0 {
			// Disable all im_subscriptions to im_topics where the user is the owner.
			sql, args, _ := sqlx.In("UPDATE im_subscriptions SET updatedat=?,deletedat=? WHERE topic IN (?)", now, now, ownTopics)
			if _, err = tx.Exec(ctx, sqlx.Rebind(sqlx.DOLLAR, sql), args...); err != nil {
				return err
			}

			// Disable group im_topics where the user is the owner.
			if _, err = tx.Exec(ctx, "UPDATE im_topics SET updatedat=$1,touchedat=$1,state=$2,stateat=$1 WHERE owner=$3",
				now, t.StateDeleted, decoded_uid); err != nil {
				return err
			}
		}

		// Disable p2p im_topics with the user (p2p topic's owner is 0).
		if _, err = tx.Exec(ctx, "UPDATE im_topics SET updatedat=$1,touchedat=$1,state=$2,stateat=$1 "+
			"FROM im_subscriptions WHERE im_topics.name=im_subscriptions.topic "+
			"AND im_topics.owner=0 AND im_subscriptions.userid=$3",
			now, t.StateDeleted, decoded_uid); err != nil {
			return err
		}

		// Disable the other user's subscription to a disabled p2p topic.
		if _, err = tx.Exec(ctx, "UPDATE im_subscriptions AS s_one SET updatedat=$1,deletedat=$1 "+
			"FROM im_subscriptions AS s_two WHERE s_one.topic=s_two.topic "+
			"AND s_two.userid=$2 AND s_two.topic LIKE 'p2p%'",
			now, decoded_uid); err != nil {
			return err
		}

		// Disable user.
		if _, err = tx.Exec(ctx, "UPDATE im_users SET updatedat=$1,state=$2,stateat=$1 WHERE id=$3",
			now, t.StateDeleted, decoded_uid); err != nil {
			return err
		}
	}

	return tx.Commit(ctx)
}

// topicStateForUser is called by UserUpdate when the update contains state change.
// Soft-deleted im_topics remain soft-deleted.
func (a *adapter) topicStateForUser(ctx context.Context, tx pgx.Tx, decoded_uid int64, now time.Time, update any) error {
	var err error

	state, ok := update.(t.ObjState)
	if !ok {
		return t.ErrMalformed
	}

	if now.IsZero() {
		now = t.TimeNow()
	}

	// Change state of all im_topics where the user is the owner.
	if _, err = tx.Exec(ctx, "UPDATE im_topics SET state=$1, stateat=$2 WHERE owner=$3 AND state!=$4",
		state, now, decoded_uid, t.StateDeleted); err != nil {
		return err
	}

	// Change state of p2p im_topics with the user (p2p topic's owner is 0)
	if _, err = tx.Exec(ctx, "UPDATE im_topics SET state=$1, stateat=$2 "+
		"FROM im_subscriptions WHERE im_topics.name=im_subscriptions.topic AND "+
		"im_topics.owner=0 AND im_subscriptions.userid=$3 AND im_topics.state!=$4",
		state, now, decoded_uid, t.StateDeleted); err != nil {
		return err
	}

	// Subscriptions don't need to be updated:
	// im_subscriptions of a disabled user are not disabled and still can be manipulated.

	return nil
}

// UserUpdate updates user object.
func (a *adapter) UserUpdate(uid t.Uid, update map[string]any) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}

	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	cols, args := common.UpdateByMap(update)
	decoded_uid := store.DecodeUid(uid)
	args = append(args, decoded_uid)
	sql, args := expandQuery("UPDATE im_users SET "+strings.Join(cols, ",")+" WHERE id=?", args...)
	_, err = tx.Exec(ctx, sql, args...)
	if err != nil {
		return err
	}

	if state, ok := update["State"]; ok {
		now, _ := update["StateAt"].(time.Time)
		err = a.topicStateForUser(ctx, tx, decoded_uid, now, state)
		if err != nil {
			return err
		}
	}

	// Tags are also stored in a separate table
	if tags := common.ExtractTags(update); tags != nil {
		// First delete all user tags
		_, err = tx.Exec(ctx, "DELETE FROM im_usertags WHERE userid=$1", decoded_uid)
		if err != nil {
			return err
		}
		// Now insert new tags
		err = addTags(ctx, tx, "im_usertags", "userid", decoded_uid, tags, false)
		if err != nil {
			return err
		}
	}

	return tx.Commit(ctx)
}

func tempFetchTags(ctx context.Context, tx pgx.Tx, decoded_uid int64) ([]string, error) {
	var allTags []string
	rows, err := tx.Query(ctx, "SELECT tag FROM im_usertags WHERE userid=$1", decoded_uid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	for rows.Next() {
		var tag string
		rows.Scan(&tag)
		allTags = append(allTags, tag)
	}
	return allTags, nil
}

// UserUpdateTags adds or resets user's tags
func (a *adapter) UserUpdateTags(uid t.Uid, add, remove, reset []string) ([]string, error) {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return nil, err
	}

	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	decoded_uid := store.DecodeUid(uid)

	if reset != nil {
		// Delete all tags first if resetting.
		_, err = tx.Exec(ctx, "DELETE FROM im_usertags WHERE userid=$1", decoded_uid)
		if err != nil {
			return nil, err
		}
		add = reset
		remove = nil
	}

	// Now insert new tags. Ignore duplicates if resetting.
	err = addTags(ctx, tx, "im_usertags", "userid", decoded_uid, add, reset == nil)
	if err != nil {
		return nil, err
	}

	// Delete tags.
	err = removeTags(ctx, tx, "im_usertags", "userid", decoded_uid, remove)
	if err != nil {
		return nil, err
	}

	var allTags []string
	rows, err := tx.Query(ctx, "SELECT tag FROM im_usertags WHERE userid=$1", decoded_uid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	for rows.Next() {
		var tag string
		rows.Scan(&tag)
		allTags = append(allTags, tag)
	}

	_, err = tx.Exec(ctx, "UPDATE im_users SET tags=$1 WHERE id=$2", t.StringSlice(allTags), decoded_uid)
	if err != nil {
		return nil, err
	}

	return allTags, tx.Commit(ctx)
}

// UserGetByCred returns user ID for the given validated credential.
func (a *adapter) UserGetByCred(method, value string) (t.Uid, error) {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	var decoded_uid int64
	err := a.db.QueryRow(ctx, "SELECT userid FROM im_credentials WHERE synthetic=$1", method+":"+value).Scan(&decoded_uid)
	if err == nil {
		return store.EncodeUid(decoded_uid), nil
	}

	if err == pgx.ErrNoRows {
		// Clear the error if user does not exist
		return t.ZeroUid, nil
	}
	return t.ZeroUid, err
}

// UserUnreadCount returns the total number of unread im_messages in all im_topics with
// the R permission. If read fails, the counts are still returned with the original
// user IDs but with the unread count undefined and non-nil error.
// UserUnreadCount does not count unread im_messages in channels although it should.
func (a *adapter) UserUnreadCount(ids ...t.Uid) (map[t.Uid]int, error) {
	uids := make([]any, len(ids))
	counts := make(map[t.Uid]int, len(ids))
	for i, id := range ids {
		uids[i] = store.DecodeUid(id)
		// Ensure all original uids are always present.
		counts[id] = 0
	}

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	// FIXME: support channels.
	query, uids := expandQuery("SELECT s.userid, SUM(t.seqid)-SUM(s.readseqid) AS unreadcount FROM im_topics AS t, im_subscriptions AS s "+
		"WHERE s.userid IN (?) AND t.name=s.topic AND s.deletedat IS NULL AND t.state!=? AND "+
		"POSITION('R' IN s.modewant)>0 AND POSITION('R' IN s.modegiven)>0 GROUP BY s.userid", uids, t.StateDeleted)
	rows, err := a.db.Query(ctx, query, uids...)
	if err != nil {
		return counts, err
	}
	defer rows.Close()

	var userId int64
	var unreadCount int
	for rows.Next() {
		if err = rows.Scan(&userId, &unreadCount); err != nil {
			break
		}
		counts[store.EncodeUid(userId)] = unreadCount
	}
	if err == nil {
		err = rows.Err()
	}

	return counts, err
}

// UserGetUnvalidated returns a list of uids which have never logged in, have no
// validated im_credentials and haven't been updated since lastUpdatedBefore.
func (a *adapter) UserGetUnvalidated(lastUpdatedBefore time.Time, limit int) ([]t.Uid, error) {
	var uids []t.Uid

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	rows, err := a.db.Query(ctx,
		"SELECT u.id, COALESCE(SUM(CASE WHEN c.done THEN 1 ELSE 0 END), 0) AS total "+
			"FROM im_users u LEFT JOIN im_credentials c ON u.id = c.userid "+
			"WHERE u.lastseen IS NULL AND u.updatedat < $1 GROUP BY u.id, u.updatedat "+
			"HAVING COALESCE(SUM(CASE WHEN c.done THEN 1 ELSE 0 END), 0) = 0 ORDER BY u.updatedat ASC LIMIT $2",
		lastUpdatedBefore, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	for rows.Next() {
		var userId int64
		var unused int
		if err = rows.Scan(&userId, &unused); err != nil {
			break
		}
		uids = append(uids, store.EncodeUid(userId))
	}
	if err == nil {
		err = rows.Err()
	}

	return uids, err
}

// *****************************

func (a *adapter) topicCreate(ctx context.Context, tx pgx.Tx, topic *t.Topic) error {
	_, err := tx.Exec(ctx, "INSERT INTO im_topics(createdat,updatedat,touchedat,state,name,usebt,owner,access,public,trusted,tags,aux) "+
		"VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)",
		topic.CreatedAt, topic.UpdatedAt, topic.TouchedAt, topic.State, topic.Id, topic.UseBt,
		store.DecodeUid(t.ParseUid(topic.Owner)), topic.Access, common.ToJSON(topic.Public), common.ToJSON(topic.Trusted),
		topic.Tags, common.ToJSON(topic.Aux))
	if err != nil {
		return err
	}

	// Save topic's tags to a separate table to make topic findable.
	return addTags(ctx, tx, "im_topictags", "topic", topic.Id, topic.Tags, false)
}

// TopicCreate saves topic object to database.
func (a *adapter) TopicCreate(topic *t.Topic) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	err = a.topicCreate(ctx, tx, topic)
	if err != nil {
		return err
	}
	return tx.Commit(ctx)
}

// If undelete = true - update subscription on duplicate key, otherwise ignore the duplicate.
func createSubscription(ctx context.Context, tx pgx.Tx, sub *t.Subscription, undelete bool) error {

	isOwner := (sub.ModeGiven & sub.ModeWant).IsOwner()

	jpriv := common.ToJSON(sub.Private)
	decoded_uid := store.DecodeUid(t.ParseUid(sub.User))
	_, err2 := tx.Exec(ctx, "SAVEPOINT createSub")
	if err2 != nil {
		log.Println("Error: Failed to create savepoint: ", err2.Error())
	}
	_, err := tx.Exec(ctx,
		"INSERT INTO im_subscriptions(createdat,updatedat,deletedat,userid,topic,modeWant,modeGiven,private) "+
			"VALUES($1,$2,NULL,$3,$4,$5,$6,$7)",
		sub.CreatedAt, sub.UpdatedAt, decoded_uid, sub.Topic, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv)

	if err != nil && isDupe(err) {
		_, err2 = tx.Exec(ctx, "ROLLBACK TO SAVEPOINT createSub")
		if err2 != nil {
			log.Println("Error: Failed to rollback savepoint: ", err2.Error())
		}
		if undelete {
			_, err = tx.Exec(ctx, "UPDATE im_subscriptions SET createdat=$1,updatedat=$2,deletedat=NULL,modeWant=$3,modeGiven=$4,"+
				"delid=0,recvseqid=0,readseqid=0 WHERE topic=$5 AND userid=$6",
				sub.CreatedAt, sub.UpdatedAt, sub.ModeWant.String(), sub.ModeGiven.String(), sub.Topic, decoded_uid)
		} else {
			_, err = tx.Exec(ctx, "UPDATE im_subscriptions SET createdat=$1,updatedat=$2,deletedat=NULL,modeWant=$3,modeGiven=$4,"+
				"delid=0,recvseqid=0,readseqid=0,private=$5 WHERE topic=$6 AND userid=$7",
				sub.CreatedAt, sub.UpdatedAt, sub.ModeWant.String(), sub.ModeGiven.String(), jpriv,
				sub.Topic, decoded_uid)
		}
	} else {
		_, err2 = tx.Exec(ctx, "RELEASE SAVEPOINT createSub")
		if err2 != nil {
			log.Println("Error: Failed to release savepoint: ", err2.Error())
		}
	}
	if err == nil && isOwner {
		_, err = tx.Exec(ctx, "UPDATE im_topics SET owner=$1 WHERE name=$2", decoded_uid, sub.Topic)
	}
	return err
}

// TopicCreateP2P given two im_users creates a p2p topic
func (a *adapter) TopicCreateP2P(initiator, invited *t.Subscription) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	err = createSubscription(ctx, tx, initiator, false)
	if err != nil {
		return err
	}

	err = createSubscription(ctx, tx, invited, true)
	if err != nil {
		return err
	}

	topic := &t.Topic{ObjHeader: t.ObjHeader{Id: initiator.Topic}}
	topic.ObjHeader.MergeTimes(&initiator.ObjHeader)
	topic.TouchedAt = initiator.GetTouchedAt()
	err = a.topicCreate(ctx, tx, topic)
	if err != nil {
		return err
	}

	return tx.Commit(ctx)
}

// TopicGet loads a single topic by name, if it exists. If the topic does not exist the call returns (nil, nil)
func (a *adapter) TopicGet(topic string) (*t.Topic, error) {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}

	// Fetch topic by name
	var tt = new(t.Topic)
	var owner int64
	err := a.db.QueryRow(ctx,
		"SELECT createdat,updatedat,state,stateat,touchedat,name AS id,usebt,access,owner,seqid,delid,subcnt,public,trusted,tags,aux "+
			"FROM im_topics WHERE name=$1",
		topic).Scan(&tt.CreatedAt, &tt.UpdatedAt, &tt.State, &tt.StateAt, &tt.TouchedAt, &tt.Id,
		&tt.UseBt, &tt.Access, &owner, &tt.SeqId, &tt.DelId, &tt.SubCnt, &tt.Public, &tt.Trusted, &tt.Tags, &tt.Aux)
	if err != nil {
		if err == pgx.ErrNoRows {
			// Nothing found - clear the error
			err = nil
		}
		return nil, err
	}

	if t.GetTopicCat(topic) == t.TopicCatGrp {
		// Topic found, get subsription count. Try both topic and channel names.
		var subCnt int
		if err = a.db.QueryRow(ctx,
			"SELECT COUNT(*) FROM im_subscriptions WHERE topic IN ($1,$2) AND deletedat IS NULL", topic, t.GrpToChn(topic)).
			Scan(&subCnt); err != nil {
			return nil, err
		}

		if subCnt != tt.SubCnt {
			// Update the topic with the correct subscription count.
			tt.SubCnt = subCnt
			if _, err = a.db.Exec(ctx, "UPDATE im_topics SET subcnt=$1 WHERE name=$2", subCnt, topic); err != nil {
				return nil, err
			}
		}
	}

	tt.Owner = store.EncodeUid(owner).String()

	return tt, err
}

// TopicsForUser loads user's contact list: p2p and grp im_topics, except for 'me' & 'fnd' im_subscriptions.
// Reads and denormalizes Public value.
func (a *adapter) TopicsForUser(uid t.Uid, keepDeleted bool, opts *t.QueryOpt) ([]t.Subscription, error) {
	// Fetch ALL user's im_subscriptions, even those which has not been modified recently.
	// We are going to use these im_subscriptions to fetch im_topics and im_users which may have been modified recently.
	q := `SELECT createdat,updatedat,deletedat,topic,delid,recvseqid,
		readseqid,modewant,modegiven,private FROM im_subscriptions WHERE userid=?`
	args := []any{store.DecodeUid(uid)}
	if !keepDeleted {
		// Filter out deleted rows.
		q += " AND deletedat IS NULL"
	}

	limit := 0
	ims := time.Time{}
	if opts != nil {
		if opts.Topic != "" {
			q += " AND topic=?"
			args = append(args, opts.Topic)
		}

		// Apply the limit only when the client does not manage the cache (or cold start).
		// Otherwise have to get all im_subscriptions and do a manual join with im_users/im_topics.
		if opts.IfModifiedSince == nil {
			if opts.Limit > 0 && opts.Limit < a.maxResults {
				limit = opts.Limit
			} else {
				limit = a.maxResults
			}
		} else {
			ims = *opts.IfModifiedSince
		}
	} else {
		limit = a.maxResults
	}

	if limit > 0 {
		q += " LIMIT ?"
		args = append(args, limit)
	}

	q, args = expandQuery(q, args...)

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	rows, err := a.db.Query(ctx, q, args...)
	if err != nil {
		return nil, err
	}
	// Must close rows manually as we will be reusing it.

	// Fetch im_subscriptions. Two queries are needed: im_users table (p2p) and im_topics table (grp).
	// Prepare a list of separate im_subscriptions to im_users vs im_topics
	join := make(map[string]t.Subscription) // Keeping these to make a join with table for .private and .access
	topq := make([]any, 0, 16)
	usrq := make([]any, 0, 16)
	for rows.Next() {
		var sub t.Subscription
		var modeWant, modeGiven []byte
		if err = rows.Scan(&sub.CreatedAt, &sub.UpdatedAt, &sub.DeletedAt, &sub.Topic, &sub.DelId,
			&sub.RecvSeqId, &sub.ReadSeqId, &modeWant, &modeGiven, &sub.Private); err != nil {
			break
		}
		sub.ModeWant.Scan(modeWant)
		sub.ModeGiven.Scan(modeGiven)
		tname := sub.Topic
		sub.User = uid.String()
		tcat := t.GetTopicCat(tname)

		if tcat == t.TopicCatMe || tcat == t.TopicCatFnd {
			// One of 'me', 'fnd' im_subscriptions, skip.
			// Don't skip 'sys' subscription.
			continue
		} else if tcat == t.TopicCatP2P {
			// P2P subscription, find the other user to get user.Public and user.Trusted.
			uid1, uid2, _ := t.ParseP2P(tname)
			if uid1 == uid {
				usrq = append(usrq, store.DecodeUid(uid2))
				sub.SetWith(uid2.UserId())
			} else {
				usrq = append(usrq, store.DecodeUid(uid1))
				sub.SetWith(uid1.UserId())
			}
		} else if tcat == t.TopicCatGrp {
			// Maybe convert channel name to topic name.
			tname = t.ChnToGrp(tname)
		}
		// No special handling needed for 'slf', 'sys' im_subscriptions.

		topq = append(topq, tname)
		sub.Private = common.FromJSON(sub.Private)
		join[tname] = sub
	}
	if err == nil {
		err = rows.Err()
	}
	rows.Close()

	if err != nil {
		return nil, err
	}

	var subs []t.Subscription
	if len(join) == 0 {
		return subs, nil
	}

	// Fetch grp im_topics and join to im_subscriptions.
	if len(topq) > 0 {
		q = "SELECT updatedat,state,touchedat,name AS id,usebt,access,seqid,delid,subcnt,public,trusted " +
			"FROM im_topics WHERE name IN (?)"
		newargs := []any{topq}

		if !keepDeleted {
			// Optionally skip deleted im_topics.
			q += " AND state!=?"
			newargs = append(newargs, t.StateDeleted)
		}

		if !ims.IsZero() {
			// Use cache timestamp if provided: get newer entries only.
			q += " AND touchedat>?"
			newargs = append(newargs, ims)

			if limit > 0 && limit < len(topq) {
				// No point in fetching more than the requested limit.
				q += " ORDER BY touchedat LIMIT ?"
				newargs = append(newargs, limit)
			}
		}
		q, newargs = expandQuery(q, newargs...)

		ctx2, cancel2 := a.getContext()
		if cancel2 != nil {
			defer cancel2()
		}
		rows, err = a.db.Query(ctx2, q, newargs...)
		if err != nil {
			return nil, err
		}

		var top t.Topic
		for rows.Next() {
			if err = rows.Scan(&top.UpdatedAt, &top.State, &top.TouchedAt, &top.Id, &top.UseBt,
				&top.Access, &top.SeqId, &top.DelId, &top.SubCnt, &top.Public, &top.Trusted); err != nil {
				break
			}

			sub := join[top.Id]
			// Check if sub.UpdatedAt needs to be adjusted to earlier or later time.
			sub.UpdatedAt = common.SelectLatestTime(sub.UpdatedAt, top.UpdatedAt)
			sub.SetState(top.State)
			sub.SetTouchedAt(top.TouchedAt)
			sub.SetSeqId(top.SeqId)
			if t.GetTopicCat(sub.Topic) == t.TopicCatGrp {
				sub.SetSubCnt(top.SubCnt)
				sub.SetPublic(top.Public)
				sub.SetTrusted(top.Trusted)
			}
			// Put back the updated value of a subsription, will process further below
			join[top.Id] = sub
		}
		if err == nil {
			err = rows.Err()
		}
		rows.Close()

		if err != nil {
			return nil, err
		}
	}

	// Fetch p2p im_users and join to p2p im_subscriptions.
	if len(usrq) > 0 {
		q = "SELECT id,updatedat,state,access,lastseen,useragent,public,trusted " +
			"FROM im_users WHERE id IN (?)"
		newargs := []any{usrq}
		if !keepDeleted {
			// Optionally skip deleted im_users.
			q += " AND state!=?"
			newargs = append(newargs, t.StateDeleted)
		}

		// Ignoring ipg: we need all im_users to get LastSeen and UserAgent.

		q, newargs = expandQuery(q, newargs...)

		ctx3, cancel3 := a.getContext()
		if cancel3 != nil {
			defer cancel3()
		}
		rows, err = a.db.Query(ctx3, q, newargs...)
		if err != nil {
			return nil, err
		}

		for rows.Next() {
			var usr2 t.User
			var id int64
			if err = rows.Scan(&id, &usr2.UpdatedAt, &usr2.State, &usr2.Access, &usr2.LastSeen, &usr2.UserAgent,
				&usr2.Public, &usr2.Trusted); err != nil {
				break
			}

			usr2.Id = store.EncodeUid(id).String()
			joinOn := uid.P2PName(t.ParseUid(usr2.Id))
			if sub, ok := join[joinOn]; ok {
				sub.UpdatedAt = common.SelectLatestTime(sub.UpdatedAt, usr2.UpdatedAt)
				sub.SetState(usr2.State)
				sub.SetPublic(usr2.Public)
				sub.SetTrusted(usr2.Trusted)
				sub.SetDefaultAccess(usr2.Access.Auth, usr2.Access.Anon)
				sub.SetLastSeenAndUA(usr2.LastSeen, usr2.UserAgent)
				join[joinOn] = sub
			}
		}
		if err == nil {
			err = rows.Err()
		}
		rows.Close()

		if err != nil {
			return nil, err
		}
	}

	subs = make([]t.Subscription, 0, len(join))
	for _, sub := range join {
		subs = append(subs, sub)
	}

	return common.SelectEarliestUpdatedSubs(subs, opts, a.maxResults), nil
}

// UsersForTopic loads im_users subscribed to the given topic.
// The difference between UsersForTopic vs SubsForTopic is that the former loads user.Public,
// the latter does not.
func (a *adapter) UsersForTopic(topic string, keepDeleted bool, opts *t.QueryOpt) ([]t.Subscription, error) {
	tcat := t.GetTopicCat(topic)

	// Fetch all subscribed im_users. The number of im_users is not large
	q := `SELECT s.createdat,s.updatedat,s.deletedat,s.userid,s.topic,s.delid,s.recvseqid,
		s.readseqid,s.modewant,s.modegiven,u.public,u.trusted,u.lastseen,u.useragent,s.private
		FROM im_subscriptions AS s JOIN im_users AS u ON s.userid=u.id
		WHERE s.topic=?`
	args := []any{topic}
	if !keepDeleted {
		// Filter out rows with im_users deleted
		q += " AND u.state!=?"
		args = append(args, t.StateDeleted)

		// For p2p im_topics we must load all im_subscriptions including deleted.
		// Otherwise it will be impossible to swipe Public values.
		if tcat != t.TopicCatP2P {
			// Filter out deleted im_subscriptions.
			q += " AND s.deletedat IS NULL"
		}
	}

	limit := a.maxResults
	var oneUser t.Uid
	if opts != nil {
		// Ignore IfModifiedSince: loading all entries because a topic cannot have too many subscribers.
		// Those unmodified will be stripped of Public & Private.

		if !opts.User.IsZero() {
			// For p2p im_topics we have to fetch both im_users otherwise public cannot be swapped.
			if tcat != t.TopicCatP2P {
				q += " AND s.userid=?"
				args = append(args, store.DecodeUid(opts.User))
			}
			oneUser = opts.User
		}
		if opts.Limit > 0 && opts.Limit < limit {
			limit = opts.Limit
		}
	}
	q += " LIMIT ?"
	args = append(args, limit)
	q, args = expandQuery(q, args...)

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	rows, err := a.db.Query(ctx, q, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	// Fetch im_subscriptions
	var sub t.Subscription
	var subs []t.Subscription
	var userId int64
	var modeWant, modeGiven []byte
	var lastSeen *time.Time = nil
	var userAgent string
	var public, trusted any
	for rows.Next() {
		if err = rows.Scan(
			&sub.CreatedAt, &sub.UpdatedAt, &sub.DeletedAt,
			&userId, &sub.Topic, &sub.DelId, &sub.RecvSeqId,
			&sub.ReadSeqId, &modeWant, &modeGiven,
			&public, &trusted, &lastSeen, &userAgent, &sub.Private); err != nil {
			break
		}

		sub.User = store.EncodeUid(userId).String()
		sub.SetPublic(public)
		sub.SetTrusted(trusted)
		sub.SetLastSeenAndUA(lastSeen, userAgent)
		sub.ModeWant.Scan(modeWant)
		sub.ModeGiven.Scan(modeGiven)
		subs = append(subs, sub)
	}
	if err == nil {
		err = rows.Err()
	}

	if err == nil && tcat == t.TopicCatP2P && len(subs) > 0 {
		// Swap public & lastSeen values of P2P im_topics as expected.
		if len(subs) == 1 {
			// The other user is deleted, nothing we can do.
			subs[0].SetPublic(nil)
			subs[0].SetTrusted(nil)
			subs[0].SetLastSeenAndUA(nil, "")
		} else {
			tmp := subs[0].GetPublic()
			subs[0].SetPublic(subs[1].GetPublic())
			subs[1].SetPublic(tmp)

			tmp = subs[0].GetTrusted()
			subs[0].SetTrusted(subs[1].GetTrusted())
			subs[1].SetTrusted(tmp)

			lastSeen := subs[0].GetLastSeen()
			userAgent = subs[0].GetUserAgent()
			subs[0].SetLastSeenAndUA(subs[1].GetLastSeen(), subs[1].GetUserAgent())
			subs[1].SetLastSeenAndUA(lastSeen, userAgent)
		}

		// Remove deleted and unneeded im_subscriptions
		if !keepDeleted || !oneUser.IsZero() {
			var xsubs []t.Subscription
			for i := range subs {
				if (subs[i].DeletedAt != nil && !keepDeleted) || (!oneUser.IsZero() && subs[i].Uid() != oneUser) {
					continue
				}
				xsubs = append(xsubs, subs[i])
			}
			subs = xsubs
		}
	}

	return subs, err
}

// topicNamesForUser reads a slice of strings using provided query.
func (a *adapter) topicNamesForUser(sqlQuery string, includeChan bool, args ...any) ([]string, error) {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	rows, err := a.db.Query(ctx, sqlQuery, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var names []string
	for rows.Next() {
		var name string
		if err = rows.Scan(&name); err != nil {
			break
		}
		names = append(names, name)
		// If the name is a group topic, also add the channel name if requested.
		if includeChan {
			if channel := t.GrpToChn(name); channel != "" {
				names = append(names, channel)
			}
		}
	}
	if err == nil {
		err = rows.Err()
	}

	return names, err
}

// OwnTopics loads a slice of topic names where the user is the owner.
func (a *adapter) OwnTopics(uid t.Uid) ([]string, error) {
	return a.topicNamesForUser("SELECT name FROM im_topics WHERE owner=$1 AND state!=$2",
		false, store.DecodeUid(uid), t.StateDeleted)
}

// ChannelsForUser loads a slice of topic names where the user is a channel reader and notifications (P) are enabled.
func (a *adapter) ChannelsForUser(uid t.Uid) ([]string, error) {
	return a.topicNamesForUser("SELECT topic FROM im_subscriptions WHERE userid=$1 AND topic LIKE 'chn%' "+
		"AND POSITION('P' IN modewant)>0 AND POSITION('P' IN modegiven)>0 AND deletedat IS NULL",
		false, store.DecodeUid(uid))
}

// TopicShare creates topic im_subscriptions and increments the topic's subcnt.
func (a *adapter) TopicShare(topic string, shares []*t.Subscription) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	for _, sub := range shares {
		err = createSubscription(ctx, tx, sub, true)
		if err != nil {
			return err
		}
	}

	if topic != "" {
		if _, err = tx.Exec(ctx, "UPDATE im_topics SET subcnt=subcnt+$1 WHERE name=$2", len(shares), topic); err != nil {
			return err
		}
	}

	return tx.Commit(ctx)
}

// TopicDelete deletes topic, im_subscriptions, im_messages.
func (a *adapter) TopicDelete(topic string, isChan, hard bool) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}

	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	// If the topic is a channel, must try to delete im_subscriptions under both grpXXX and chnXXX names.
	args := []any{topic}
	if isChan {
		args = append(args, t.GrpToChn(topic))
	}

	if hard {
		// Delete im_subscriptions. If this is a channel, delete both group im_subscriptions and channel im_subscriptions.
		q, args := expandQuery("DELETE FROM im_subscriptions WHERE topic IN (?)", args)
		if _, err = tx.Exec(ctx, q, args...); err != nil {
			return err
		}

		if err = messageDeleteList(ctx, tx, topic, nil); err != nil {
			return err
		}

		if _, err = tx.Exec(ctx, "DELETE FROM im_topictags WHERE topic=$1", topic); err != nil {
			return err
		}

		if _, err = tx.Exec(ctx, "DELETE FROM im_topics WHERE name=$1", topic); err != nil {
			return err
		}
	} else {
		now := t.TimeNow()

		q, args := expandQuery("UPDATE im_subscriptions SET updatedat=?,deletedat=? WHERE topic IN (?)", now, now, args)
		if _, err = tx.Exec(ctx, q, args...); err != nil {
			return err
		}

		if _, err = tx.Exec(ctx, "UPDATE im_topics SET updatedat=$1,touchedat=$1,state=$2,stateat=$1 WHERE name=$3",
			now, t.StateDeleted, topic); err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

func (a *adapter) TopicUpdateOnMessage(topic string, msg *t.Message) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	_, err := a.db.Exec(ctx, "UPDATE im_topics SET seqid=$1,touchedat=$2 WHERE name=$3", msg.SeqId, msg.CreatedAt, topic)

	return err
}

// TopicUpdateSubCnt updates subscriber count denormalized in topic.
func (a *adapter) TopicUpdateSubCnt(topic string) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	_, err := a.db.Exec(ctx,
		"UPDATE im_topics SET subcnt=(SELECT COUNT(*) FROM im_subscriptions WHERE topic IN ($1,$2) AND deletedat IS NULL) WHERE name=$1",
		topic, t.GrpToChn(topic))
	return err
}

func (a *adapter) TopicUpdate(topic string, update map[string]any) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}

	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	if t, u := update["TouchedAt"], update["UpdatedAt"]; t == nil && u != nil {
		update["TouchedAt"] = u
	}
	cols, args := common.UpdateByMap(update)
	q, args := expandQuery("UPDATE im_topics SET "+strings.Join(cols, ",")+" WHERE name=?", args, topic)
	_, err = tx.Exec(ctx, q, args...)
	if err != nil {
		return err
	}

	// Tags are also stored in a separate table
	if tags := common.ExtractTags(update); tags != nil {
		// First delete all user tags
		_, err = tx.Exec(ctx, "DELETE FROM im_topictags WHERE topic=$1", topic)
		if err != nil {
			return err
		}
		// Now insert new tags
		err = addTags(ctx, tx, "im_topictags", "topic", topic, tags, false)
		if err != nil {
			return err
		}
	}

	return tx.Commit(ctx)
}

func (a *adapter) TopicOwnerChange(topic string, newOwner t.Uid) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	_, err := a.db.Exec(ctx, "UPDATE im_topics SET owner=$1 WHERE name=$2", store.DecodeUid(newOwner), topic)
	return err
}

// Get a subscription of a user to a topic.
func (a *adapter) SubscriptionGet(topic string, user t.Uid, keepDeleted bool) (*t.Subscription, error) {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	query := `SELECT createdat,updatedat,deletedat,userid AS user,topic,delid,recvseqid,
		readseqid,modewant,modegiven,private FROM im_subscriptions WHERE topic=$1 AND userid=$2`
	if !keepDeleted {
		query += " AND deletedat IS NULL"
	}
	var sub t.Subscription
	var userId int64
	var modeWant, modeGiven []byte
	err := a.db.QueryRow(ctx, query, topic, store.DecodeUid(user)).Scan(&sub.CreatedAt, &sub.UpdatedAt, &sub.DeletedAt, &userId,
		&sub.Topic, &sub.DelId, &sub.RecvSeqId, &sub.ReadSeqId, &modeWant, &modeGiven, &sub.Private)

	if err != nil {
		if err == pgx.ErrNoRows {
			// Nothing found - clear the error
			err = nil
		}
		return nil, err
	}

	sub.User = store.EncodeUid(userId).String()
	sub.ModeWant.Scan(modeWant)
	sub.ModeGiven.Scan(modeGiven)

	return &sub, nil
}

// SubsForUser loads all user's im_subscriptions. Does NOT load Public or Private values and does
// not load deleted im_subscriptions.
func (a *adapter) SubsForUser(forUser t.Uid) ([]t.Subscription, error) {
	q := `SELECT createdat,updatedat,deletedat,userid AS user,topic,delid,recvseqid,
		readseqid,modewant,modegiven FROM im_subscriptions WHERE userid=$1 AND deletedat IS NULL`
	args := []any{store.DecodeUid(forUser)}

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	rows, err := a.db.Query(ctx, q, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var subs []t.Subscription
	var sub t.Subscription
	var userId int64
	var modeWant, modeGiven []byte
	for rows.Next() {
		if err = rows.Scan(&sub.CreatedAt, &sub.UpdatedAt, &sub.DeletedAt, &userId, &sub.Topic, &sub.DelId,
			&sub.RecvSeqId, &sub.ReadSeqId, &modeWant, &modeGiven); err != nil {
			break
		}

		sub.User = store.EncodeUid(userId).String()
		sub.ModeWant.Scan(modeWant)
		sub.ModeGiven.Scan(modeGiven)
		subs = append(subs, sub)
	}
	if err == nil {
		err = rows.Err()
	}

	return subs, err
}

// SubsForTopic fetches all subsciptions for a topic. Does NOT load Public value.
// The difference between UsersForTopic vs SubsForTopic is that the former loads user.public+trusted,
// the latter does not.
func (a *adapter) SubsForTopic(topic string, keepDeleted bool, opts *t.QueryOpt) ([]t.Subscription, error) {
	q := `SELECT createdat,updatedat,deletedat,userid AS user,topic,delid,recvseqid,
		readseqid,modewant,modegiven,private FROM im_subscriptions WHERE topic=?`

	args := []any{topic}
	if !keepDeleted {
		// Filter out deleted rows.
		q += " AND deletedat IS NULL"
	}
	limit := a.maxResults
	if opts != nil {
		// Ignore IfModifiedSince - we must return all entries
		// Those unmodified will be stripped of Public & Private.

		if !opts.User.IsZero() {
			q += " AND userid=?"
			args = append(args, store.DecodeUid(opts.User))
		}
		if opts.Limit > 0 && opts.Limit < limit {
			limit = opts.Limit
		}
	}

	q += " LIMIT ?"
	args = append(args, limit)
	q, args = expandQuery(q, args...)

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	rows, err := a.db.Query(ctx, q, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var subs []t.Subscription
	var sub t.Subscription
	var userId int64
	var modeWant, modeGiven []byte
	for rows.Next() {
		if err = rows.Scan(&sub.CreatedAt, &sub.UpdatedAt, &sub.DeletedAt, &userId, &sub.Topic, &sub.DelId,
			&sub.RecvSeqId, &sub.ReadSeqId, &modeWant, &modeGiven, &sub.Private); err != nil {
			break
		}

		sub.User = store.EncodeUid(userId).String()
		sub.ModeWant.Scan(modeWant)
		sub.ModeGiven.Scan(modeGiven)
		subs = append(subs, sub)
	}
	if err == nil {
		err = rows.Err()
	}

	return subs, err
}

// SubsUpdate updates one or multiple im_subscriptions to a topic.
func (a *adapter) SubsUpdate(topic string, user t.Uid, update map[string]any) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}

	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	cols, args := common.UpdateByMap(update)
	q := "UPDATE im_subscriptions SET " + strings.Join(cols, ",") + " WHERE topic=?"
	args = append(args, topic)
	if !user.IsZero() {
		// Update just one topic subscription
		q += " AND userid=?"
		args = append(args, store.DecodeUid(user))
	}
	q, args = expandQuery(q, args...)

	if _, err = tx.Exec(ctx, q, args...); err != nil {
		return err
	}

	return tx.Commit(ctx)
}

// SubsDelete marks at most one subscription as deleted.
func (a *adapter) SubsDelete(topic string, user t.Uid) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	tx, err := a.db.Begin(ctx)
	if err != nil {
		return err
	}

	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	decoded_id := store.DecodeUid(user)
	now := t.TimeNow()
	res, err := tx.Exec(ctx,
		"UPDATE im_subscriptions SET updatedat=$1,deletedat=$2 WHERE topic=$3 AND userid=$4 AND deletedat IS NULL",
		now, now, topic, decoded_id)
	if err != nil {
		return err
	}

	affected := res.RowsAffected()
	if affected == 0 {
		// ensure tx.Rollback() above is ran
		err = t.ErrNotFound
		return err
	}

	// Channel readers cannot delete im_messages.
	if !t.IsChannel(topic) {
		// Remove records of im_messages soft-deleted by this user.
		_, err = tx.Exec(ctx, "DELETE FROM im_dellog WHERE topic=$1 AND deletedfor=$2", topic, decoded_id)
		if err != nil {
			return err
		}
	}

	if t.GetTopicCat(topic) == t.TopicCatGrp {
		// Decrement topic subscription count (only one subscription is	deleted).
		_, err = tx.Exec(ctx, "UPDATE im_topics SET subcnt=subcnt-1 WHERE name=$1", topic)
		if err != nil {
			return err
		}
	}

	return tx.Commit(ctx)
}

// subsDelForUser marks user's im_subscriptions as deleted.
func subsDelForUser(ctx context.Context, tx pgx.Tx, decoded_uid int64, hard bool) error {
	// Decrement subscription count for all im_topics the user is subscribed to.
	rows, err := tx.Query(ctx, "SELECT topic FROM im_subscriptions WHERE userid=$1 AND deletedat IS NULL", decoded_uid)
	if err != nil {
		return err
	}
	var im_topics []any
	for rows.Next() {
		var name string
		if err = rows.Scan(&name); err != nil {
			break
		}
		if t.IsChannel(name) {
			// Convert channel name to group name.
			name = t.ChnToGrp(name)
		}
		im_topics = append(im_topics, name)
	}
	if err == nil {
		err = rows.Err()
	}
	rows.Close()
	if err != nil {
		return err
	}
	if len(im_topics) > 0 {
		sql, args, _ := sqlx.In("UPDATE im_topics SET subcnt=subcnt-1 WHERE name IN (?)", im_topics)
		_, err = tx.Exec(ctx, sqlx.Rebind(sqlx.DOLLAR, sql), args...)
		if err != nil {
			return err
		}
	}

	if hard {
		// Hard delete: remove all im_subscriptions for the user.
		_, err = tx.Exec(ctx, "DELETE FROM im_subscriptions WHERE userid=$1", decoded_uid)
	} else {
		now := t.TimeNow()
		_, err = tx.Exec(ctx, "UPDATE im_subscriptions SET updatedat=$1,deletedat=$2 WHERE userid=$3 AND deletedat IS NULL;",
			now, now, decoded_uid)
	}
	return err
}

// SubsDelForUser marks user's im_subscriptions as deleted.
func (a *adapter) SubsDelForUser(user t.Uid, hard bool) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}

	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}

	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	if err = subsDelForUser(ctx, tx, store.DecodeUid(user), hard); err != nil {
		return err
	}

	return tx.Commit(ctx)

}

// Find returns a list of im_users and group im_topics which match the given tags, such as "email:jdoe@example.com" or "tel:+18003287448".
func (a *adapter) Find(caller, promoPrefix string, req [][]string, opt []string, activeOnly bool) ([]t.Subscription, error) {
	index := make(map[string]struct{})
	var args []any
	constraint := ""
	allReq := t.FlattenDoubleSlice(req)
	for _, tag := range append(allReq, opt...) {
		args = append(args, tag)
		index[tag] = struct{}{}
	}
	if len(args) == 0 {
		// Nothing to search for.
		return nil, nil
	}
	constraint += "tg.tag IN (?) "
	constraint, args, err := sqlx.In(constraint, args)
	if err != nil {
		return nil, err
	}
	if activeOnly {
		args = append(args, t.StateOK)
		constraint += "AND state=? "
	}
	constraint = sqlx.Rebind(sqlx.DOLLAR, constraint)

	var matcher string
	if promoPrefix != "" {
		// The max number of tags is 16. Using 20 to make sure one prefix match is greater than all non-prefix matches.
		matcher = "SUM(CASE WHEN POSITION('" + promoPrefix + "' IN tg.tag)=1 THEN 20 ELSE 1 END)"
	} else {
		matcher = "COUNT(*)"
	}

	query := "SELECT CAST(u.id AS VARCHAR) AS topic,u.createdat,u.updatedat,FALSE,u.access::jsonb,0 AS subcnt,u.public::jsonb,u.trusted::jsonb,u.tags::jsonb," +
		matcher + " AS matches " +
		"FROM im_users AS u JOIN im_usertags AS tg ON tg.userid=u.id " +
		"WHERE " + constraint +
		"GROUP BY u.id,u.createdat,u.updatedat,u.access::jsonb,u.public::jsonb,u.trusted::jsonb,u.tags::jsonb "

	having := ""
	if len(allReq) > 0 {
		var a []any
		having, a = common.DisjunctionSql(req, "tg.tag")
		having = rebindWithStart(having, len(args)+1)
		query += having
		args = append(args, a...)
	}

	query += "UNION ALL "

	query += "SELECT t.name AS topic,t.createdat,t.updatedat,t.usebt,t.access::jsonb,t.subcnt,t.public::jsonb,t.trusted::jsonb,t.tags::jsonb," +
		matcher + " AS matches " +
		"FROM im_topics AS t JOIN im_topictags AS tg ON t.name=tg.topic " +
		"WHERE " + constraint +
		"GROUP BY t.name,t.createdat,t.updatedat,t.usebt,t.access::jsonb,t.subcnt,t.public::jsonb,t.trusted::jsonb,t.tags::jsonb "
	if having != "" {
		query += having
	}
	args = append(args, a.maxResults)
	query += "ORDER BY matches DESC, subcnt DESC LIMIT $" + strconv.Itoa(len(args))

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	// Get im_users matched by tags, sort by number of matches from high to low.
	rows, err := a.db.Query(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	// Fetch im_subscriptions
	var public, trusted any
	var access t.DefaultAccess
	var subcnt int
	var setTags t.StringSlice
	var ignored int
	var isChan bool
	var sub t.Subscription
	var subs []t.Subscription
	for rows.Next() {
		if err = rows.Scan(&sub.Topic, &sub.CreatedAt, &sub.UpdatedAt, &isChan, &access, &subcnt,
			&public, &trusted, &setTags, &ignored); err != nil {
			subs = nil
			break
		}

		if id, err := strconv.ParseInt(sub.Topic, 10, 64); err == nil {
			sub.Topic = store.EncodeUid(id).UserId()
			if sub.Topic == caller {
				// Skip the caller.
				continue
			}
		}

		if isChan {
			// This is a channel, convert grp to chn name.
			sub.Topic = t.GrpToChn(sub.Topic)
		}

		sub.SetSubCnt(subcnt)
		sub.SetPublic(public)
		sub.SetTrusted(trusted)
		sub.SetDefaultAccess(access.Auth, access.Anon)
		// Indicating that the mode is not set, not 'N'.
		sub.ModeGiven = t.ModeUnset
		sub.ModeWant = t.ModeUnset
		sub.Private = common.FilterFoundTags(setTags, index)
		subs = append(subs, sub)
	}

	if err == nil {
		err = rows.Err()
	}

	return subs, err

}

// FindOne returns topic or user which matches the given tag.
func (a *adapter) FindOne(tag string) (string, error) {
	var args []any
	query := "SELECT t.name AS topic FROM im_topics AS t LEFT JOIN im_topictags AS tt ON t.name=tt.topic " +
		"WHERE tt.tag=?"
	args = append(args, tag)

	query += " UNION ALL "

	query += "SELECT CAST(u.id AS VARCHAR) AS topic FROM im_users AS u LEFT JOIN im_usertags AS ut ON ut.userid=u.id " +
		"WHERE ut.tag=?"
	args = append(args, tag)

	// LIMIT is applied to all resultant rows.
	query += " LIMIT 1"

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	query, args = expandQuery(query, args)
	rows, err := a.db.Query(ctx, query, args...)
	if err != nil {
		return "", err
	}
	defer rows.Close()

	var found string
	if rows.Next() {
		if err = rows.Scan(&found); err != nil {
			return "", err
		}

		// Check if the found value is a topic name or a user ID.
		// User IDs are returned as decoded decimal strings.
		if id, err := strconv.ParseInt(found, 10, 64); err == nil {
			found = store.EncodeUid(id).UserId()
		}
	}
	if err == nil {
		err = rows.Err()
	}

	return found, err
}

// Messages
func (a *adapter) MessageSave(msg *t.Message) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	// store assignes message ID, but we don't use it. Message IDs are not used anywhere.
	// Using a sequential ID provided by the database.
	var id int
	err := a.db.QueryRow(ctx,
		`INSERT INTO im_messages(createdAt,updatedAt,seqid,topic,"from",head,content) VALUES($1,$2,$3,$4,$5,$6,$7) RETURNING id`,
		msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic,
		store.DecodeUid(t.ParseUid(msg.From)), msg.Head, common.ToJSON(msg.Content)).Scan(&id)
	if err == nil {
		// Replacing ID given by store by ID given by the DB.
		msg.SetUid(t.Uid(id))
	}
	return err
}

func (a *adapter) MessageGetAll(topic string, forUser t.Uid, opts *t.QueryOpt) ([]t.Message, error) {
	var limit = a.maxMessageResults

	args := []any{store.DecodeUid(forUser), topic}
	seqIdConstraint := ""
	if opts != nil {
		seqIdConstraint = "AND m.seqid "
		if len(opts.IdRanges) > 0 {
			constr, newargs := common.RangesToSql(opts.IdRanges)
			seqIdConstraint += constr
			args = append(args, newargs...)
		} else {
			seqIdConstraint += "BETWEEN ? AND ?"
			if opts.Since > 0 {
				args = append(args, opts.Since)
			} else {
				args = append(args, 0)
			}
			if opts.Before > 0 {
				// BETWEEN is inclusive-inclusive, Tinode API requires inclusive-exclusive, thus -1
				args = append(args, opts.Before-1)
			} else {
				args = append(args, 1<<31-1)
			}
		}

		if opts.Limit > 0 && opts.Limit < limit {
			limit = opts.Limit
		}
	}

	args = append(args, limit)

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	query, args := expandQuery(`SELECT m.createdat,m.updatedat,m.deletedat,m.delid,m.seqid,m.topic,m."from",m.head,m.content`+
		" FROM im_messages AS m LEFT JOIN im_dellog AS d"+
		" ON d.topic=m.topic AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=?"+
		" WHERE m.delid=0 AND m.topic=? "+seqIdConstraint+" AND d.deletedfor IS NULL"+
		" ORDER BY m.seqid DESC LIMIT ?", args...)
	rows, err := a.db.Query(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	msgs := make([]t.Message, 0, limit)
	for rows.Next() {
		var msg t.Message
		var from int64
		if err = rows.Scan(&msg.CreatedAt, &msg.UpdatedAt, &msg.DeletedAt, &msg.DelId, &msg.SeqId,
			&msg.Topic, &from, &msg.Head, &msg.Content); err != nil {
			break
		}
		msg.From = store.EncodeUid(from).String()
		msgs = append(msgs, msg)
	}
	if err == nil {
		err = rows.Err()
	}

	return msgs, err
}

// Get ranges of deleted im_messages
func (a *adapter) MessageGetDeleted(topic string, forUser t.Uid, opts *t.QueryOpt) ([]t.DelMessage, error) {
	var limit = a.maxResults
	var lower = 0
	var upper = 1<<31 - 1

	if opts != nil {
		if opts.Since > 0 {
			lower = opts.Since
		}
		if opts.Before > 1 {
			// DelRange is inclusive-exclusive, while BETWEEN is inclusive-inclisive.
			upper = opts.Before - 1
		}

		if opts.Limit > 0 && opts.Limit < limit {
			limit = opts.Limit
		}
	}

	// Fetch log of deletions
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	rows, err := a.db.Query(ctx, "SELECT topic,deletedfor,delid,low,hi FROM im_dellog WHERE topic=$1 AND delid BETWEEN $2 AND $3"+
		" AND (deletedFor=0 OR deletedFor=$4) ORDER BY delid LIMIT $5",
		topic, lower, upper, store.DecodeUid(forUser), limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var im_dellog struct {
		Topic      string
		Deletedfor int64
		Delid      int
		Low        int
		Hi         int
	}
	var dmsgs []t.DelMessage
	var dmsg t.DelMessage
	for rows.Next() {
		if err = rows.Scan(&im_dellog.Topic, &im_dellog.Deletedfor, &im_dellog.Delid, &im_dellog.Low, &im_dellog.Hi); err != nil {
			dmsgs = nil
			break
		}

		if im_dellog.Delid != dmsg.DelId {
			if dmsg.DelId > 0 {
				dmsgs = append(dmsgs, dmsg)
			}
			dmsg.DelId = im_dellog.Delid
			dmsg.Topic = im_dellog.Topic
			if im_dellog.Deletedfor > 0 {
				dmsg.DeletedFor = store.EncodeUid(im_dellog.Deletedfor).String()
			} else {
				dmsg.DeletedFor = ""
			}
			dmsg.SeqIdRanges = nil
		}
		if im_dellog.Hi <= im_dellog.Low+1 {
			im_dellog.Hi = 0
		}
		dmsg.SeqIdRanges = append(dmsg.SeqIdRanges, t.Range{Low: im_dellog.Low, Hi: im_dellog.Hi})
	}
	if err == nil {
		err = rows.Err()
	}

	if err == nil {
		if dmsg.DelId > 0 {
			dmsgs = append(dmsgs, dmsg)
		}
	}

	return dmsgs, err
}

func messageDeleteList(ctx context.Context, tx pgx.Tx, topic string, toDel *t.DelMessage) error {
	var err error

	if toDel == nil {
		// Whole topic is being deleted, thus also deleting all im_messages.
		_, err = tx.Exec(ctx, "DELETE FROM im_dellog WHERE topic=$1", topic)
		if err == nil {
			_, err = tx.Exec(ctx, "DELETE FROM im_messages WHERE topic=$1", topic)
		}
		// im_filemsglinks will be deleted because of ON DELETE CASCADE
		return err
	}

	// Only some im_messages are being deleted

	delRanges := toDel.SeqIdRanges

	if toDel.DeletedFor == "" {
		// Hard-deleting im_messages requires updates to the im_messages table.
		where := "m.topic=? "
		args := []any{topic}

		if len(delRanges) > 0 {
			rSql, rArgs := common.RangesToSql(delRanges)
			where += " AND m.seqid " + rSql
			args = append(args, rArgs...)
		}

		where += " AND m.deletedat IS NULL"

		// We are asked to delete im_messages no older than newerThan.
		if newerThan := toDel.GetNewerThan(); newerThan != nil {
			where += " AND m.createdat>?"
			args = append(args, newerThan)
		}

		// Find the actual IDs still present in the database.
		var seqIDs []int
		query, newargs := expandQuery("SELECT seqid FROM im_messages AS m WHERE "+where, args)
		rows, err := tx.Query(ctx, query, newargs...)
		if err != nil {
			return err
		}
		defer rows.Close()

		for rows.Next() {
			var seqID int
			if err := rows.Scan(&seqID); err != nil {
				return err
			}
			seqIDs = append(seqIDs, seqID)
		}
		if err = rows.Err(); err != nil {
			return err
		}

		if len(seqIDs) == 0 {
			// Nothing to delete. No need to make a log entry. All done.
			return nil
		}

		// Recalculate the actual ranges to delete.
		sort.Ints(seqIDs)
		delRanges = t.SliceToRanges(seqIDs)

		// Compose a new query with the new ranges.
		where = "m.topic=?"
		args = []any{topic}
		rSql, rArgs := common.RangesToSql(delRanges)
		where += " AND m.seqid " + rSql
		args = append(args, rArgs...)

		// No need to add anything else: deletedat etc is already accounted for.

		query, newargs = expandQuery("DELETE FROM im_filemsglinks AS fml USING im_messages AS m WHERE m.id=fml.msgid AND "+
			where, args...)
		_, err = tx.Exec(ctx, query, newargs...)
		if err != nil {
			return err
		}

		query, newargs = expandQuery(`UPDATE im_messages AS m SET deletedat=?,delid=?,"from"=0,head=NULL,content=NULL WHERE `+
			where, t.TimeNow(), toDel.DelId, args)
		_, err = tx.Exec(ctx, query, newargs...)
		if err != nil {
			return err
		}
	}

	// Now make log entries. Needed for both hard- and soft-deleting.

	// Prepare statement is not needed because the driver prepares the statement on first use then caches it.
	forUser := common.DecodeUidString(toDel.DeletedFor)
	for _, rng := range toDel.SeqIdRanges {
		if rng.Hi == 0 {
			// Dellog must contain valid Low and *Hi*.
			rng.Hi = rng.Low + 1
		}

		if _, err = tx.Exec(ctx, "INSERT INTO im_dellog(topic,deletedfor,delid,low,hi) VALUES($1,$2,$3,$4,$5)",
			topic, forUser, toDel.DelId, rng.Low, rng.Hi); err != nil {
			break
		}
	}

	return err
}

// MessageDeleteList deletes im_messages in the given topic with seqIds from the list.
func (a *adapter) MessageDeleteList(topic string, toDel *t.DelMessage) (err error) {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}

	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	if err = messageDeleteList(ctx, tx, topic, toDel); err != nil {
		return err
	}

	return tx.Commit(ctx)
}

func deviceHasher(deviceID string) string {
	// Generate custom key as [64-bit hash of device id] to ensure predictable
	// length of the key
	hasher := fnv.New64()
	hasher.Write([]byte(deviceID))
	return strconv.FormatUint(uint64(hasher.Sum64()), 16)
}

// Device management for push notifications
func (a *adapter) DeviceUpsert(uid t.Uid, def *t.DeviceDef) error {
	hash := deviceHasher(def.DeviceId)

	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	// Ensure uniqueness of the device ID: delete all records of the device ID
	_, err = tx.Exec(ctx, "DELETE FROM im_devices WHERE hash=$1", hash)
	if err != nil {
		return err
	}

	// Actually add/update DeviceId for the new user
	_, err = tx.Exec(ctx, "INSERT INTO im_devices(userid, hash, deviceId, platform, lastseen, lang) VALUES($1,$2,$3,$4,$5,$6)",
		store.DecodeUid(uid), hash, def.DeviceId, def.Platform, def.LastSeen, def.Lang)
	if err != nil {
		return err
	}

	return tx.Commit(ctx)
}

func (a *adapter) DeviceGetAll(uids ...t.Uid) (map[t.Uid][]t.DeviceDef, int, error) {
	var unums []any
	for _, uid := range uids {
		unums = append(unums, store.DecodeUid(uid))
	}

	query, unums := expandQuery("SELECT userid,deviceid,platform,lastseen,lang FROM im_devices WHERE userid IN (?)", unums)
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	rows, err := a.db.Query(ctx, query, unums...)
	if err != nil {
		return nil, 0, err
	}
	defer rows.Close()

	var device struct {
		Userid   int64
		Deviceid string
		Platform string
		Lastseen time.Time
		Lang     string
	}

	result := make(map[t.Uid][]t.DeviceDef)
	count := 0
	for rows.Next() {
		if err = rows.Scan(&device.Userid, &device.Deviceid, &device.Platform, &device.Lastseen, &device.Lang); err != nil {
			break
		}
		uid := store.EncodeUid(device.Userid)
		udev := result[uid]
		udev = append(udev, t.DeviceDef{
			DeviceId: device.Deviceid,
			Platform: device.Platform,
			LastSeen: device.Lastseen,
			Lang:     device.Lang,
		})
		result[uid] = udev
		count++
	}
	if err == nil {
		err = rows.Err()
	}

	return result, count, err
}

func deviceDelete(ctx context.Context, tx pgx.Tx, uid t.Uid, deviceID string) error {
	var err error
	var res pgconn.CommandTag
	if deviceID == "" {
		res, err = tx.Exec(ctx, "DELETE FROM im_devices WHERE userid=$1", store.DecodeUid(uid))
	} else {
		res, err = tx.Exec(ctx, "DELETE FROM im_devices WHERE userid=$1 AND hash=$2", store.DecodeUid(uid), deviceHasher(deviceID))
	}

	if err == nil {
		if count := res.RowsAffected(); count == 0 {
			err = t.ErrNotFound
		}
	}

	return err
}

func (a *adapter) DeviceDelete(uid t.Uid, deviceID string) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	err = deviceDelete(ctx, tx, uid, deviceID)
	if err != nil {
		return err
	}

	return tx.Commit(ctx)
}

// Credential management

// CredUpsert adds or updates a validation record. Returns true if inserted, false if updated.
// 1. if credential is validated:
// 1.1 Hard-delete unconfirmed equivalent record, if exists.
// 1.2 Insert new. Report error if duplicate.
// 2. if credential is not validated:
// 2.1 Check if validated equivalent exist. If so, report an error.
// 2.2 Soft-delete all unvalidated records of the same method.
// 2.3 Undelete existing credential. Return if successful.
// 2.4 Insert new credential record.
func (a *adapter) CredUpsert(cred *t.Credential) (bool, error) {
	var err error

	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return false, err
	}
	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	now := t.TimeNow()
	userId := common.DecodeUidString(cred.User)

	// Enforce uniqueness: if credential is confirmed, "method:value" must be unique.
	// if credential is not yet confirmed, "userid:method:value" is unique.
	synth := cred.Method + ":" + cred.Value

	if !cred.Done {
		// Check if this credential is already validated.
		var done bool
		err = tx.QueryRow(ctx, "SELECT done FROM im_credentials WHERE synthetic=$1", synth).Scan(&done)
		if err == nil {
			// Assign err to ensure closing of a transaction.
			err = t.ErrDuplicate
			return false, err
		}
		if err != pgx.ErrNoRows {
			return false, err
		}
		// We are going to insert new record.
		synth = cred.User + ":" + synth

		// Adding new unvalidated credential. Deactivate all unvalidated records of this user and method.
		_, err = tx.Exec(ctx, "UPDATE im_credentials SET deletedat=$1 WHERE userid=$2 AND method=$3 AND done=FALSE",
			now, userId, cred.Method)
		if err != nil {
			return false, err
		}
		// Assume that the record exists and try to update it: undelete, update timestamp and response value.
		res, err := tx.Exec(ctx, "UPDATE im_credentials SET updatedat=$1,deletedat=NULL,resp=$2,done=FALSE WHERE synthetic=$3",
			cred.UpdatedAt, cred.Resp, synth)
		if err != nil {
			return false, err
		}
		// If record was updated, then all is fine.
		if numrows := res.RowsAffected(); numrows > 0 {
			return false, tx.Commit(ctx)
		}
	} else {
		// Hard-deleting unconformed record if it exists.
		_, err = tx.Exec(ctx, "DELETE FROM im_credentials WHERE synthetic=$1", cred.User+":"+synth)
		if err != nil {
			return false, err
		}
	}

	_, err = tx.Exec(ctx, "INSERT INTO im_credentials(createdat,updatedat,method,value,synthetic,userid,resp,done) "+
		"VALUES($1,$2,$3,$4,$5,$6,$7,$8)",
		cred.CreatedAt, cred.UpdatedAt, cred.Method, cred.Value, synth, userId, cred.Resp, cred.Done)
	if err != nil {
		if isDupe(err) {
			return true, t.ErrDuplicate
		}
		return true, err
	}
	return true, tx.Commit(ctx)
}

// credDel deletes given validation method or all methods of the given user.
// 1. If user is being deleted, hard-delete all records (method == "")
// 2. If one value is being deleted:
// 2.1 Delete it if it's valiated or if there were no attempts at validation
// (otherwise it could be used to circumvent the limit on validation attempts).
// 2.2 In that case mark it as soft-deleted.
func credDel(ctx context.Context, tx pgx.Tx, uid t.Uid, method, value string) error {
	constraints := " WHERE userid=?"
	args := []any{store.DecodeUid(uid)}

	if method != "" {
		constraints += " AND method=?"
		args = append(args, method)

		if value != "" {
			constraints += " AND value=?"
			args = append(args, value)
		}
	}
	where, _ := expandQuery(constraints, args...)

	var err error
	var res pgconn.CommandTag
	if method == "" {
		// Case 1
		res, err = tx.Exec(ctx, "DELETE FROM im_credentials"+where, args...)
		if err == nil {
			if count := res.RowsAffected(); count == 0 {
				err = t.ErrNotFound
			}
		}
		return err
	}

	// Case 2.1
	res, err = tx.Exec(ctx, "DELETE FROM im_credentials"+where+" AND (done=TRUE OR retries=0)", args...)
	if err != nil {
		return err
	}
	if count := res.RowsAffected(); count > 0 {
		return nil
	}

	// Case 2.2
	query, args := expandQuery("UPDATE im_credentials SET deletedat=?"+constraints, t.TimeNow(), args)
	res, err = tx.Exec(ctx, query, args...)
	if err == nil {
		if count := res.RowsAffected(); count >= 0 {
			err = t.ErrNotFound
		}
	}

	return err
}

// CredDel deletes either im_credentials of the given user. If method is blank all
// im_credentials are removed. If value is blank all im_credentials of the given the
// method are removed.
func (a *adapter) CredDel(uid t.Uid, method, value string) error {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	err = credDel(ctx, tx, uid, method, value)
	if err != nil {
		return err
	}

	return tx.Commit(ctx)
}

// CredConfirm marks given credential method as confirmed.
func (a *adapter) CredConfirm(uid t.Uid, method string) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	res, err := a.db.Exec(
		ctx,
		"UPDATE im_credentials SET updatedat=$1,done=TRUE,synthetic=CONCAT(method,':',value) "+
			"WHERE userid=$2 AND method=$3 AND deletedat IS NULL AND done=FALSE",
		t.TimeNow(), store.DecodeUid(uid), method)
	if err != nil {
		if isDupe(err) {
			return t.ErrDuplicate
		}
		return err
	}
	if numrows := res.RowsAffected(); numrows < 1 {
		return t.ErrNotFound
	}
	return nil
}

// CredFail increments failure count of the given validation method.
func (a *adapter) CredFail(uid t.Uid, method string) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	_, err := a.db.Exec(ctx, "UPDATE im_credentials SET updatedat=$1,retries=retries+1 WHERE userid=$2 AND method=$3 AND done=FALSE",
		t.TimeNow(), store.DecodeUid(uid), method)
	return err
}

// CredGetActive returns currently active unvalidated credential of the given user and method.
func (a *adapter) CredGetActive(uid t.Uid, method string) (*t.Credential, error) {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	var cred t.Credential

	err := a.db.QueryRow(ctx, "SELECT createdat,updatedat,method,value,resp,done,retries "+
		"FROM im_credentials WHERE userid=$1 AND deletedat IS NULL AND method=$2 AND done=FALSE",
		store.DecodeUid(uid), method).Scan(&cred.CreatedAt, &cred.UpdatedAt, &cred.Method, &cred.Value, &cred.Resp, &cred.Done, &cred.Retries)
	if err != nil {
		if err == pgx.ErrNoRows {
			err = nil
		}
		return nil, err
	}
	cred.User = uid.String()

	return &cred, nil
}

// CredGetAll returns credential records for the given user and method, all or validated only.
func (a *adapter) CredGetAll(uid t.Uid, method string, validatedOnly bool) ([]t.Credential, error) {
	query := "SELECT createdat,updatedat,method,value,resp,done,retries FROM im_credentials WHERE userid=$1 AND deletedat IS NULL"
	args := []any{store.DecodeUid(uid)}
	if method != "" {
		query += " AND method=$2"
		args = append(args, method)
	}
	if validatedOnly {
		query += " AND done=TRUE"
	}

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	var im_credentials []t.Credential
	rows, err := a.db.Query(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	for rows.Next() {
		var cred t.Credential
		if err = rows.Scan(&cred.CreatedAt, &cred.UpdatedAt, &cred.Method, &cred.Value, &cred.Resp, &cred.Done, &cred.Retries); err != nil {
			im_credentials = nil
			break
		}

		im_credentials = append(im_credentials, cred)
	}

	user := uid.String()
	for i := range im_credentials {
		im_credentials[i].User = user
	}

	return im_credentials, err
}

// FileUploads

// FileStartUpload initializes a file upload
func (a *adapter) FileStartUpload(fd *t.FileDef) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	var user any
	if fd.User != "" {
		user = store.DecodeUid(t.ParseUid(fd.User))
	}
	_, err := a.db.Exec(ctx,
		"INSERT INTO im_fileuploads(id,createdat,updatedat,userid,status,mimetype,size,etag,location) "+
			"VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)",
		store.DecodeUid(fd.Uid()), fd.CreatedAt, fd.UpdatedAt, user,
		fd.Status, fd.MimeType, fd.Size, fd.ETag, fd.Location)
	return err
}

// FileFinishUpload marks file upload as completed, successfully or otherwise
func (a *adapter) FileFinishUpload(fd *t.FileDef, success bool, size int64) (*t.FileDef, error) {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return nil, err
	}
	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	now := t.TimeNow()
	if success {
		_, err = tx.Exec(ctx, "UPDATE im_fileuploads SET updatedat=$1,status=$2,size=$3,etag=$4,location=$5 WHERE id=$6",
			now, t.UploadCompleted, size, fd.ETag, fd.Location, store.DecodeUid(fd.Uid()))
		if err != nil {
			return nil, err
		}

		fd.Status = t.UploadCompleted
		fd.Size = size
	} else {
		// Deleting the record: there is no value in keeping it in the DB.
		_, err = tx.Exec(ctx, "DELETE FROM im_fileuploads WHERE id=$1", store.DecodeUid(fd.Uid()))
		if err != nil {
			return nil, err
		}

		fd.Status = t.UploadFailed
		fd.Size = 0
	}
	fd.UpdatedAt = now

	return fd, tx.Commit(ctx)
}

// FileGet fetches a record of a specific file
func (a *adapter) FileGet(fid string) (*t.FileDef, error) {
	id := t.ParseUid(fid)
	if id.IsZero() {
		return nil, t.ErrMalformed
	}

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}
	var fd t.FileDef
	var ID int64
	var userId int64
	err := a.db.QueryRow(ctx, "SELECT id,createdat,updatedat,userid AS user,status,mimetype,size,etag,location "+
		"FROM im_fileuploads WHERE id=$1", store.DecodeUid(id)).Scan(&ID, &fd.CreatedAt, &fd.UpdatedAt, &userId, &fd.Status,
		&fd.MimeType, &fd.Size, &fd.ETag, &fd.Location)
	if err == pgx.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}

	fd.Id = common.EncodeUidString(fd.Id).String()
	fd.User = store.EncodeUid(userId).String()

	return &fd, nil
}

// FileDeleteUnused deletes file upload records.
func (a *adapter) FileDeleteUnused(olderThan time.Time, limit int) ([]string, error) {
	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return nil, err
	}
	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	// Garbage collecting entries which as either marked as deleted, or lack message references, or have no user assigned.
	query := "SELECT fu.id,fu.location FROM im_fileuploads AS fu LEFT JOIN im_filemsglinks AS fml ON fml.fileid=fu.id " +
		"WHERE fml.id IS NULL"
	var args []any
	if !olderThan.IsZero() {
		query += " AND fu.updatedat<?"
		args = append(args, olderThan)
	}
	if limit > 0 {
		query += " LIMIT ?"
		args = append(args, limit)
	}
	query, _ = expandQuery(query, args...)

	rows, err := tx.Query(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var locations []string
	var ids []any
	for rows.Next() {
		var id int
		var loc string
		if err = rows.Scan(&id, &loc); err != nil {
			break
		}
		if loc != "" {
			locations = append(locations, loc)
		}
		ids = append(ids, id)
	}
	if err == nil {
		err = rows.Err()
	}

	if err != nil {
		return nil, err
	}

	if len(ids) > 0 {
		query, ids = expandQuery("DELETE FROM im_fileuploads WHERE id IN (?)", ids)
		_, err = tx.Exec(ctx, query, ids...)
		if err != nil {
			return nil, err
		}
	}

	return locations, tx.Commit(ctx)
}

// FileLinkAttachments connects given topic or message to the file record IDs from the list.
func (a *adapter) FileLinkAttachments(topic string, userId, msgId t.Uid, fids []string) error {
	if len(fids) == 0 || (topic == "" && msgId.IsZero() && userId.IsZero()) {
		return t.ErrMalformed
	}
	now := t.TimeNow()

	var args []any
	var linkId any
	var linkBy string
	if !msgId.IsZero() {
		linkBy = "msgid"
		linkId = int64(msgId)
	} else if topic != "" {
		linkBy = "topic"
		linkId = topic
		// Only one attachment per topic is permitted at this time.
		fids = fids[0:1]
	} else {
		linkBy = "userid"
		linkId = store.DecodeUid(userId)
		// Only one attachment per user is permitted at this time.
		fids = fids[0:1]
	}

	// Decoded ids
	var dids []any
	for _, fid := range fids {
		id := t.ParseUid(fid)
		if id.IsZero() {
			return t.ErrMalformed
		}
		dids = append(dids, store.DecodeUid(id))
	}

	for _, id := range dids {
		// createdat,fileid,[msgid|topic|userid]
		args = append(args, now, id, linkId)
	}

	ctx, cancel := a.getContextForTx()
	if cancel != nil {
		defer cancel()
	}
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			tx.Rollback(ctx)
		}
	}()

	// Unlink earlier uploads on the same topic or user allowing them to be garbage-collected.
	if msgId.IsZero() {
		sql := "DELETE FROM im_filemsglinks WHERE " + linkBy + "=$1"
		_, err = tx.Exec(ctx, sql, linkId)
		if err != nil {
			return err
		}
	}

	query, args := expandQuery("INSERT INTO im_filemsglinks(createdat,fileid,"+linkBy+") VALUES (?,?,?)"+
		strings.Repeat(",(?,?,?)", len(dids)-1), args...)
	_, err = tx.Exec(ctx, query, args...)
	if err != nil {
		return err
	}

	return tx.Commit(ctx)
}

// PCacheGet reads a persistet cache entry.
func (a *adapter) PCacheGet(key string) (string, error) {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	var value string
	if err := a.db.QueryRow(ctx, `SELECT "value" FROM im_kvmeta WHERE "key"=$1 LIMIT 1`, key).Scan(&value); err != nil {
		if err == pgx.ErrNoRows {
			return "", t.ErrNotFound
		}
		return "", err
	}
	return value, nil
}

// PCacheUpsert creates or updates a persistent cache entry.
func (a *adapter) PCacheUpsert(key string, value string, failOnDuplicate bool) error {
	if strings.Contains(key, "%") {
		// Do not allow % in keys: it interferes with LIKE query.
		return t.ErrMalformed
	}

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	var action string
	if !failOnDuplicate {
		action = ` ON CONFLICT ("key") DO UPDATE SET createdat=$2,"value"=$3`
	}

	_, err := a.db.Exec(ctx, `INSERT INTO im_kvmeta("key",createdat,"value") VALUES($1,$2,$3)`+action,
		key, t.TimeNow(), value)
	if isDupe(err) {
		return t.ErrDuplicate
	}
	return err
}

// PCacheDelete deletes one persistent cache entry.
func (a *adapter) PCacheDelete(key string) error {
	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	_, err := a.db.Exec(ctx, `DELETE FROM im_kvmeta WHERE "key"=$1`, key)
	return err
}

// PCacheExpire expires old entries with the given key prefix.
func (a *adapter) PCacheExpire(keyPrefix string, olderThan time.Time) error {
	if keyPrefix == "" {
		return t.ErrMalformed
	}

	ctx, cancel := a.getContext()
	if cancel != nil {
		defer cancel()
	}

	_, err := a.db.Exec(ctx, `DELETE FROM im_kvmeta WHERE "key" LIKE $1 AND createdat<$2`, keyPrefix+"%", olderThan)
	return err
}

// GetTestDB returns a currently open database connection.
func (a *adapter) GetTestDB() any {
	return a.db
}

// Helper functions

// Check if MySQL error is a Error Code: 1062. Duplicate entry ... for key ...
func isDupe(err error) bool {
	if err == nil {
		return false
	}

	msg := err.Error()
	return strings.Contains(msg, "SQLSTATE 23505")
}

func isMissingTable(err error) bool {
	if err == nil {
		return false
	}

	msg := err.Error()
	return strings.Contains(msg, "SQLSTATE 42P01")
}

func isMissingDb(err error) bool {
	if err == nil {
		return false
	}

	msg := err.Error()
	return strings.Contains(msg, "SQLSTATE 3D000")
}

// setConnStr converts a config structure to a DSN connection string.
func setConnStr(c configType) (string, error) {
	// Default to disable SSL mode.
	sslMode := "disable"
	if c.SSLMode != "" {
		sslMode = c.SSLMode
	}

	if c.User == "" || c.Passwd == "" || c.Host == "" || c.Port == "" || c.DBName == "" {
		return "", errors.New("adapter postgres invalid config value")
	}
	connStr := fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=%s&connect_timeout=%d",
		c.User,
		c.Passwd,
		c.Host,
		c.Port,
		c.DBName,
		sslMode,
		c.SqlTimeout)

	return connStr, nil
}

// expandQuery replaces the placeholders in the query with the actual values and returns
// the expanded query and the arguments to be used in the query.
func expandQuery(query string, args ...any) (string, []any) {
	var expandedArgs []any
	var expandedQuery string

	if len(args) != strings.Count(query, "?") {
		args = flattenSlice(args)
	}
	expandedQuery, expandedArgs, _ = sqlx.In(query, args...)
	return sqlx.Rebind(sqlx.DOLLAR, expandedQuery), expandedArgs
}

// flatMap converts a slice of mixed values/slices into a flat slice.
func flattenSlice(slice []any) []any {
	var result []any
	for _, v := range slice {
		switch reflect.TypeOf(v).Kind() {
		case reflect.Slice:
			s := reflect.ValueOf(v)
			for i := 0; i < s.Len(); i++ {
				result = append(result, s.Index(i).Interface())
			}
		default:
			result = append(result, v)
		}
	}
	return result
}

// Rebind a query from ? to the target $ with custom initial value.
func rebindWithStart(query string, startAt int) string {
	// Add space enough for 10 params before we have to allocate
	rqb := make([]byte, 0, len(query)+10)

	var i, j = 0, startAt

	for i = strings.Index(query, "?"); i != -1; i = strings.Index(query, "?") {
		rqb = append(rqb, query[:i]...)
		rqb = append(rqb, '$')

		rqb = strconv.AppendInt(rqb, int64(j), 10)
		j++

		query = query[i+1:]
	}

	return string(append(rqb, query...))
}

// GetTestAdapter returns an adapter object. Useful for running tests.
func GetTestAdapter() *adapter {
	return &adapter{}
}

func init() {
	store.RegisterAdapter(&adapter{})
}