"""Database layer (SQLAlchemy 2.0 async). PostgreSQL in production, SQLite for local tests."""
from __future__ import annotations

import asyncio
import logging
import os
from datetime import datetime, timezone
from typing import Any, Optional
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

from sqlalchemy import (
    JSON,
    BigInteger,
    Boolean,
    DateTime,
    ForeignKey,
    Integer,
    String,
    Text,
    UniqueConstraint,
    delete,
    desc,
    func,
    select,
    update,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.pool import NullPool

log = logging.getLogger(__name__)

JSONType = JSON().with_variant(JSONB(), "postgresql")


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


class Base(DeclarativeBase):
    pass


class Channel(Base):
    __tablename__ = "channels"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False)
    access_hash: Mapped[int] = mapped_column(BigInteger, default=0)
    title: Mapped[str] = mapped_column(String(256), default="")
    username: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
    active: Mapped[bool] = mapped_column(Boolean, default=True)
    added_by: Mapped[int] = mapped_column(BigInteger, default=0)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)


class Post(Base):
    __tablename__ = "posts"
    __table_args__ = (UniqueConstraint("channel_id", "message_id", name="uq_post_channel_msg"),)

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    channel_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("channels.id"), index=True)
    message_id: Mapped[Optional[int]] = mapped_column(BigInteger, nullable=True)
    status: Mapped[str] = mapped_column(String(16), default="draft", index=True)  # draft | sent
    source: Mapped[str] = mapped_column(String(16), default="bot")  # bot | adopted
    text: Mapped[str] = mapped_column(Text, default="")
    entities: Mapped[list] = mapped_column(JSONType, default=list)
    media_kind: Mapped[Optional[str]] = mapped_column(String(16), nullable=True)
    media_file_id: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    buttons: Mapped[list] = mapped_column(JSONType, default=list)
    link_preview: Mapped[bool] = mapped_column(Boolean, default=False)
    created_by: Mapped[int] = mapped_column(BigInteger, default=0)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    sent_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)

    def as_dict(self) -> dict:
        return {c.name: getattr(self, c.name) for c in self.__table__.columns}


class Batch(Base):
    """One /swap run (so it can be undone)."""

    __tablename__ = "batches"

    id: Mapped[str] = mapped_column(String(32), primary_key=True)
    old_username: Mapped[str] = mapped_column(String(64))
    new_username: Mapped[str] = mapped_column(String(64))
    created_by: Mapped[int] = mapped_column(BigInteger, default=0)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    undone: Mapped[bool] = mapped_column(Boolean, default=False)


class Change(Base):
    __tablename__ = "changes"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    batch_id: Mapped[str] = mapped_column(String(32), ForeignKey("batches.id"), index=True)
    channel_id: Mapped[int] = mapped_column(BigInteger)
    message_id: Mapped[int] = mapped_column(BigInteger)
    old_state: Mapped[dict] = mapped_column(JSONType, default=dict)
    new_state: Mapped[dict] = mapped_column(JSONType, default=dict)
    reverted: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)


class Migration(Base):
    """One /repost run: a channel copied post by post, so it can be continued, undone or finished."""

    __tablename__ = "migrations"

    id: Mapped[str] = mapped_column(String(32), primary_key=True)
    channel_id: Mapped[int] = mapped_column(BigInteger, index=True)
    old_username: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
    new_username: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
    include_typed: Mapped[bool] = mapped_column(Boolean, default=True)
    include_posts: Mapped[bool] = mapped_column(Boolean, default=False)
    first_id: Mapped[int] = mapped_column(BigInteger)
    last_id: Mapped[int] = mapped_column(BigInteger)
    partial: Mapped[bool] = mapped_column(Boolean, default=False)
    # copying | stopped | incomplete | copied | old_deleted | copies_deleted
    status: Mapped[str] = mapped_column(String(16), default="copying")
    created_by: Mapped[int] = mapped_column(BigInteger, default=0)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)


class MigrationItem(Base):
    __tablename__ = "migration_items"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    migration_id: Mapped[str] = mapped_column(String(32), ForeignKey("migrations.id"), index=True)
    old_id: Mapped[int] = mapped_column(BigInteger)
    new_id: Mapped[int] = mapped_column(BigInteger)
    old_deleted: Mapped[bool] = mapped_column(Boolean, default=False)
    new_deleted: Mapped[bool] = mapped_column(Boolean, default=False)


class Shift(Base):
    """One /shift run: posts of a source channel copied into another channel (so it can be continued or undone)."""

    __tablename__ = "shifts"

    id: Mapped[str] = mapped_column(String(32), primary_key=True)
    src_channel_id: Mapped[int] = mapped_column(BigInteger, index=True)
    src_access_hash: Mapped[int] = mapped_column(BigInteger, default=0)
    src_title: Mapped[str] = mapped_column(String(256), default="")
    src_username: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
    dst_channel_id: Mapped[int] = mapped_column(BigInteger, index=True)
    first_id: Mapped[int] = mapped_column(BigInteger)
    last_id: Mapped[int] = mapped_column(BigInteger)
    # copying | stopped | incomplete | copied | copies_deleted
    status: Mapped[str] = mapped_column(String(16), default="copying")
    created_by: Mapped[int] = mapped_column(BigInteger, default=0)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)


class ShiftItem(Base):
    __tablename__ = "shift_items"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    shift_id: Mapped[str] = mapped_column(String(32), ForeignKey("shifts.id"), index=True)
    old_id: Mapped[int] = mapped_column(BigInteger)  # message id in the source channel
    new_id: Mapped[int] = mapped_column(BigInteger)  # message id of the copy in the destination channel
    new_deleted: Mapped[bool] = mapped_column(Boolean, default=False)


class ScheduledDelete(Base):
    """A channel message the bot has to delete at a given time (a broadcast with an expiry)."""

    __tablename__ = "scheduled_deletes"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    channel_id: Mapped[int] = mapped_column(BigInteger, index=True)
    message_id: Mapped[int] = mapped_column(BigInteger)
    delete_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
    group_id: Mapped[str] = mapped_column(String(32), default="")  # the broadcast it belongs to
    status: Mapped[str] = mapped_column(String(12), default="pending")  # pending | failed
    attempts: Mapped[int] = mapped_column(Integer, default=0)
    last_error: Mapped[Optional[str]] = mapped_column(String(160), nullable=True)
    created_by: Mapped[int] = mapped_column(BigInteger, default=0)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)


class ChannelSync(Base):
    """Where the last look at a channel stopped, so the next one only has to read what is new."""

    __tablename__ = "channel_sync"

    channel_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False)
    last_top: Mapped[int] = mapped_column(BigInteger, default=0)  # the newest message id that existed then
    last_pts: Mapped[int] = mapped_column(BigInteger, default=0)  # the channel's event counter then
    deferred: Mapped[int] = mapped_column(Integer, default=0)  # new messages left for the next look (too fresh then)
    last_run: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    last_deep: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)  # last look at every id


class ChannelHistory(Base):
    """How far the older posts of a channel (the ones sent before the bot was there) have been read into My posts."""

    __tablename__ = "channel_history"

    channel_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False)
    top: Mapped[int] = mapped_column(BigInteger, default=0)  # the newest message id that counts as "older"
    next_id: Mapped[int] = mapped_column(BigInteger, default=1)  # where the reading goes on
    imported: Mapped[int] = mapped_column(Integer, default=0)  # older posts added to My posts so far
    done: Mapped[bool] = mapped_column(Boolean, default=False)
    started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    finished_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)


class IgnoredPost(Base):
    """A post the owner made the bot forget: it stays in the channel, but the sync never takes it over again."""

    __tablename__ = "ignored_posts"
    __table_args__ = (UniqueConstraint("channel_id", "message_id", name="uq_ignored_channel_msg"),)

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    channel_id: Mapped[int] = mapped_column(BigInteger, index=True)
    message_id: Mapped[int] = mapped_column(BigInteger)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)


class JobMark(Base):
    """A fact about a /repost or /shift run that has to outlive a restart, e.g. that links in other channels were
    pointed at its copies (so that Undo points them back), or which extras a shift was asked for."""

    __tablename__ = "job_marks"

    job_id: Mapped[str] = mapped_column(String(32), primary_key=True)
    mark: Mapped[str] = mapped_column(String(24), primary_key=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)


class Admin(Base):
    __tablename__ = "admins"
    user_id: Mapped[int] = mapped_column(BigInteger, primary_key=True)


def as_utc(dt: Optional[datetime]) -> Optional[datetime]:
    """A datetime from the database as an aware UTC one (SQLite hands back naive values)."""
    if dt is None:
        return None
    return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt.astimezone(timezone.utc)


def normalize_db_url(url: str) -> tuple:
    """Turn a plain DATABASE_URL into a SQLAlchemy async URL + connect args."""
    connect_args: dict = {}
    if url.startswith("postgres://"):
        url = "postgresql://" + url[len("postgres://"):]
    if url.startswith("postgresql://"):
        url = "postgresql+asyncpg://" + url[len("postgresql://"):]
    if url.startswith("postgresql+asyncpg://"):
        parts = urlsplit(url)
        keep = []
        for k, v in parse_qsl(parts.query, keep_blank_values=True):
            if k == "sslmode":
                if v in ("require", "verify-ca", "verify-full"):
                    connect_args["ssl"] = True
            elif k in ("channel_binding", "options"):
                continue
            else:
                keep.append((k, v))
        host = parts.hostname or ""
        if "-pooler" in host or "pgbouncer" in host:
            # PgBouncer in transaction mode does not support prepared statements
            connect_args["statement_cache_size"] = 0
            keep.append(("prepared_statement_cache_size", "0"))
        url = urlunsplit(parts._replace(query=urlencode(keep)))
    elif url.startswith("sqlite:///"):
        url = "sqlite+aiosqlite:///" + url[len("sqlite:///"):]
    return url, connect_args


class Database:
    def __init__(self, url: str, pool: str = "null"):
        self.url, connect_args = normalize_db_url(url)
        kwargs: dict = {"connect_args": connect_args}
        if self.url.startswith("sqlite"):
            path = self.url.split(":///", 1)[-1]
            if path and path != ":memory:":
                os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
        elif pool == "queue":
            kwargs.update(pool_size=5, max_overflow=5, pool_pre_ping=True, pool_recycle=1800)
        else:
            # short-lived connections: lets serverless Postgres (e.g. Neon) go to sleep when idle
            kwargs["poolclass"] = NullPool
        self.engine = create_async_engine(self.url, **kwargs)
        self.Session = async_sessionmaker(self.engine, expire_on_commit=False)

    async def init(self, retries: int = 6) -> None:
        last: Optional[BaseException] = None
        for i in range(retries):
            try:
                async with self.engine.begin() as conn:
                    await conn.run_sync(Base.metadata.create_all)
                return
            except Exception as e:  # cold start of serverless DBs can fail once
                last = e
                log.warning("database not ready (%s) - retry %s/%s", e, i + 1, retries)
                await asyncio.sleep(min(2 ** i, 15))
        assert last is not None
        raise last

    async def close(self) -> None:
        await self.engine.dispose()

    async def list_admins(self) -> set[int]:
        async with self.Session() as s:
            return set((await s.execute(select(Admin.user_id))).scalars())

    async def add_admin(self, uid: int) -> bool:
        async with self.Session() as s:
            if await s.get(Admin, uid) is not None:
                return False
            s.add(Admin(user_id=uid))
            await s.commit()
            return True

    async def remove_admin(self, uid: int) -> bool:
        async with self.Session() as s:
            row = await s.get(Admin, uid)
            if row is None:
                return False
            await s.delete(row)
            await s.commit()
            return True

    # ------------------------------------------------------------------ channels
    async def list_channels(self, active_only: bool = True) -> list:
        async with self.Session() as s:
            q = select(Channel).order_by(Channel.title)
            if active_only:
                q = q.where(Channel.active.is_(True))
            return list((await s.execute(q)).scalars())

    async def get_channel(self, cid: int) -> Optional[Channel]:
        async with self.Session() as s:
            return await s.get(Channel, cid)

    async def save_channel(self, cid: int, access_hash: int, title: str, username: Optional[str], added_by: int) -> Channel:
        async with self.Session() as s:
            ch = await s.get(Channel, cid)
            if ch is None:
                ch = Channel(id=cid, access_hash=access_hash, title=title, username=username, added_by=added_by, active=True)
                s.add(ch)
            else:
                ch.access_hash = access_hash or ch.access_hash
                ch.title = title or ch.title
                ch.username = username
                ch.active = True
            await s.commit()
            return ch

    async def touch_channel_hash(self, cid: int, access_hash: int) -> None:
        async with self.Session() as s:
            ch = await s.get(Channel, cid)
            if ch is not None and access_hash and ch.access_hash != access_hash:
                ch.access_hash = access_hash
                await s.commit()

    async def set_channel_active(self, cid: int, active: bool) -> None:
        async with self.Session() as s:
            ch = await s.get(Channel, cid)
            if ch is not None:
                ch.active = active
                await s.commit()

    async def count_channel_posts(self, cid: int) -> int:
        """Every post of the channel in My posts - published ones and drafts."""
        async with self.Session() as s:
            return int((await s.execute(select(func.count()).select_from(Post).where(Post.channel_id == cid))).scalar_one())

    async def delete_channel_data(self, cid: int) -> int:
        """Everything My posts holds about a channel: its saved posts (published and drafts) and what the sync keeps
        about them - the posts forgotten on purpose, where the last look and the reading of the older posts stopped - so a
        channel that is added again starts from scratch. Nothing is touched in Telegram. The channel itself stays
        registered (switched off) because reposts and shifts that name it still need it, and so do timed deletions.
        Returns how many posts went."""
        async with self.Session() as s:
            n = (await s.execute(select(func.count()).select_from(Post).where(Post.channel_id == cid))).scalar_one()
            await s.execute(delete(Post).where(Post.channel_id == cid))
            for model in (IgnoredPost, ChannelSync, ChannelHistory):
                await s.execute(delete(model).where(model.channel_id == cid))
            await s.commit()
            return int(n)

    # --------------------------------------------------------------------- posts
    async def create_post(self, **fields: Any) -> Post:
        async with self.Session() as s:
            p = Post(**fields)
            s.add(p)
            await s.commit()
            return p

    async def get_post(self, pid: int) -> Optional[Post]:
        async with self.Session() as s:
            return await s.get(Post, pid)

    async def update_post(self, pid: int, **fields: Any) -> Optional[Post]:
        async with self.Session() as s:
            p = await s.get(Post, pid)
            if p is None:
                return None
            for k, v in fields.items():
                setattr(p, k, v)
            p.updated_at = utcnow()
            await s.commit()
            return p

    async def delete_post(self, pid: int) -> None:
        async with self.Session() as s:
            p = await s.get(Post, pid)
            if p is not None:
                await s.delete(p)
                await s.commit()

    async def list_posts(self, channel_id: Optional[int], status: Optional[str], offset: int, limit: int) -> tuple:
        cond = []
        if channel_id is not None:
            cond.append(Post.channel_id == channel_id)
        if status:
            cond.append(Post.status == status)
        async with self.Session() as s:
            total = (await s.execute(select(func.count()).select_from(Post).where(*cond))).scalar_one()
            q = (
                select(Post)
                .where(*cond)
                .order_by(Post.status.asc(), desc(Post.message_id), desc(Post.id))
                .offset(offset)
                .limit(limit)
            )
            rows = list((await s.execute(q)).scalars())
            return rows, total

    async def latest_sent(self, channel_id: int, limit: int = 30) -> list:
        async with self.Session() as s:
            q = (
                select(Post)
                .where(Post.channel_id == channel_id, Post.status == "sent", Post.message_id.is_not(None))
                .order_by(desc(Post.message_id))
                .limit(limit)
            )
            return list((await s.execute(q)).scalars())

    async def sent_posts(self, channel_id: int) -> list:
        """Every published post of a channel, oldest first."""
        async with self.Session() as s:
            q = (
                select(Post)
                .where(Post.channel_id == channel_id, Post.status == "sent", Post.message_id.is_not(None))
                .order_by(Post.message_id)
            )
            return list((await s.execute(q)).scalars())

    async def sent_posts_from(self, channel_id: int, first_message_id: int) -> list:
        """Published posts of a channel from message id `first_message_id` on, oldest first."""
        async with self.Session() as s:
            q = (
                select(Post)
                .where(
                    Post.channel_id == channel_id,
                    Post.status == "sent",
                    Post.message_id.is_not(None),
                    Post.message_id >= first_message_id,
                )
                .order_by(Post.message_id, Post.id)
            )
            return list((await s.execute(q)).scalars())

    async def mine_message_ids(self, channel_id: int) -> set:
        async with self.Session() as s:
            q = select(Post.message_id).where(
                Post.channel_id == channel_id, Post.source == "bot", Post.message_id.is_not(None)
            )
            return {r for r in (await s.execute(q)).scalars()}

    async def find_by_message(self, channel_id: int, message_id: int) -> Optional[Post]:
        async with self.Session() as s:
            q = select(Post).where(Post.channel_id == channel_id, Post.message_id == message_id)
            return (await s.execute(q)).scalar_one_or_none()

    async def forget_posts(self, channel_id: int, message_ids: list) -> int:
        """Remove the saved posts of these channel messages (they were deleted in the channel) and what hangs on them:
        a timed deletion that is not needed any more, a "forgotten on purpose" mark. Returns how many posts went."""
        if not message_ids:
            return 0
        ids = list(message_ids)
        async with self.Session() as s:
            rows = (await s.execute(select(Post).where(Post.channel_id == channel_id, Post.message_id.in_(ids)))).scalars().all()
            for p in rows:
                await s.delete(p)
            for model in (ScheduledDelete, IgnoredPost):
                extra = await s.execute(select(model).where(model.channel_id == channel_id, model.message_id.in_(ids)))
                for r in extra.scalars().all():
                    await s.delete(r)
            await s.commit()
            return len(rows)

    async def posts_at(self, channel_id: int, message_ids: list) -> list:
        """The published posts that sit at these message ids."""
        if not message_ids:
            return []
        async with self.Session() as s:
            q = select(Post).where(
                Post.channel_id == channel_id, Post.status == "sent", Post.message_id.in_(list(message_ids))
            )
            return list((await s.execute(q)).scalars())

    async def release_slot(self, channel_id: int, message_id: int, keep_pid: Optional[int] = None) -> int:
        """Drop any other saved post at this message id (the sync may have picked the message up a moment earlier)."""
        async with self.Session() as s:
            q = select(Post).where(Post.channel_id == channel_id, Post.message_id == message_id)
            if keep_pid is not None:
                q = q.where(Post.id != keep_pid)
            rows = list((await s.execute(q)).scalars())
            for p in rows:
                await s.delete(p)
            if rows:
                await s.commit()
            return len(rows)

    async def _drop_slots(self, s, channel_id: int, message_ids: list) -> None:
        """Inside a transaction: remove saved posts at these ids, so the posts about to be saved can take their place."""
        q = select(Post).where(Post.channel_id == channel_id, Post.message_id.in_(list(message_ids)))
        rows = list((await s.execute(q)).scalars())
        for p in rows:
            await s.delete(p)
        if rows:
            await s.flush()

    # ------------------------------------------------------- what happens outside the bot
    async def get_sync(self, channel_id: int) -> Optional[ChannelSync]:
        async with self.Session() as s:
            return await s.get(ChannelSync, channel_id)

    async def max_saved_id(self, channel_id: int) -> int:
        """The newest message id among the channel's published posts (0 if there is none)."""
        async with self.Session() as s:
            q = select(func.max(Post.message_id)).where(
                Post.channel_id == channel_id, Post.status == "sent", Post.message_id.is_not(None)
            )
            return int((await s.execute(q)).scalar_one_or_none() or 0)

    async def save_sync(
        self, channel_id: int, *, last_top: int, last_pts: int, deferred: int = 0, deep: bool = False
    ) -> None:
        async with self.Session() as s:
            row = await s.get(ChannelSync, channel_id)
            if row is None:
                row = ChannelSync(channel_id=channel_id)
                s.add(row)
            now = utcnow()
            row.last_top, row.last_pts, row.deferred, row.last_run = last_top, last_pts, deferred, now
            if deep:
                row.last_deep = now
            await s.commit()

    async def adopt_message(self, channel_id: int, message_id: int, fields: dict, sent_at: Optional[datetime] = None) -> tuple:
        """Save a post somebody else made in the channel. Returns (post, True), or (None, False) when the bot already
        has a post at that message id (also when another task saved it a moment ago)."""
        async with self.Session() as s:
            q = select(Post).where(Post.channel_id == channel_id, Post.message_id == message_id)
            if (await s.execute(q)).scalar_one_or_none() is not None:
                return None, False
            p = Post(channel_id=channel_id, message_id=message_id, status="sent", source="adopted", created_by=0,
                     sent_at=sent_at or utcnow(), **fields)
            s.add(p)
            try:
                await s.commit()
            except IntegrityError:
                await s.rollback()
                return None, False
            return p, True

    async def adopt_many(self, channel_id: int, items: list) -> list:
        """items: [(message id, post fields, sent_at)] - save the posts somebody else made, many at once. Returns the
        message ids that were new (the bot already had a post at the others)."""
        if not items:
            return []
        ids = [i[0] for i in items]
        async with self.Session() as s:
            have = set(
                (await s.execute(select(Post.message_id).where(Post.channel_id == channel_id, Post.message_id.in_(ids)))).scalars()
            )
            fresh = [i for i in items if i[0] not in have]
            if not fresh:
                return []
            s.add_all(
                [
                    Post(channel_id=channel_id, message_id=mid, status="sent", source="adopted", created_by=0,
                         sent_at=at or utcnow(), **fields)
                    for mid, fields, at in fresh
                ]
            )
            try:
                await s.commit()
                return [i[0] for i in fresh]
            except IntegrityError:  # somebody saved one of them a moment ago: one by one then
                await s.rollback()
        made = []
        for mid, fields, at in fresh:
            _, created = await self.adopt_message(channel_id, mid, fields, sent_at=at)
            if created:
                made.append(mid)
        return made

    # ------------------------------------------------------------- the channel's older posts
    async def get_history(self, channel_id: int) -> Optional[ChannelHistory]:
        async with self.Session() as s:
            return await s.get(ChannelHistory, channel_id)

    async def start_history(self, channel_id: int, top: int) -> ChannelHistory:
        """Note that the older posts of a channel (message ids 1 to `top`) are to be read in. Does nothing when that
        has begun already."""
        async with self.Session() as s:
            row = await s.get(ChannelHistory, channel_id)
            if row is not None:
                return row
            row = ChannelHistory(channel_id=channel_id, top=max(0, top), next_id=1, imported=0)
            if top < 1:  # an empty channel: nothing to read
                row.done, row.finished_at = True, utcnow()
            s.add(row)
            try:
                await s.commit()
            except IntegrityError:
                await s.rollback()
                row = await s.get(ChannelHistory, channel_id)
            return row

    async def save_history(self, channel_id: int, *, next_id: int, imported: int = 0, done: bool = False) -> None:
        async with self.Session() as s:
            row = await s.get(ChannelHistory, channel_id)
            if row is None:
                return
            row.next_id = max(row.next_id, next_id)
            row.imported += imported
            if done and not row.done:
                row.done, row.finished_at = True, utcnow()
            await s.commit()

    async def reset_history(self, channel_id: int) -> None:
        """Read the channel's older posts again at the next look (the posts the bot already has are skipped)."""
        async with self.Session() as s:
            row = await s.get(ChannelHistory, channel_id)
            if row is not None:
                await s.delete(row)
                await s.commit()

    # ------------------------------------------------------------ posts forgotten on purpose
    async def ignore_post(self, channel_id: int, message_id: int) -> None:
        async with self.Session() as s:
            q = select(IgnoredPost).where(IgnoredPost.channel_id == channel_id, IgnoredPost.message_id == message_id)
            if (await s.execute(q)).scalar_one_or_none() is not None:
                return
            s.add(IgnoredPost(channel_id=channel_id, message_id=message_id))
            try:
                await s.commit()
            except IntegrityError:
                await s.rollback()

    async def ignored_ids(self, channel_id: int) -> set:
        async with self.Session() as s:
            q = select(IgnoredPost.message_id).where(IgnoredPost.channel_id == channel_id)
            return {int(r) for r in (await s.execute(q)).scalars()}

    # ------------------------------------------------------------ facts about a repost / shift run
    async def set_mark(self, job_id: str, mark: str) -> None:
        async with self.Session() as s:
            if await s.get(JobMark, (job_id, mark)) is not None:
                return
            s.add(JobMark(job_id=job_id, mark=mark))
            try:
                await s.commit()
            except IntegrityError:
                await s.rollback()

    async def has_mark(self, job_id: str, mark: str) -> bool:
        async with self.Session() as s:
            return await s.get(JobMark, (job_id, mark)) is not None

    async def marks_of(self, job_id: str) -> set:
        async with self.Session() as s:
            q = select(JobMark.mark).where(JobMark.job_id == job_id)
            return set((await s.execute(q)).scalars())

    async def sync_update_post(self, pid: int, expected: Optional[datetime], fields: dict) -> bool:
        """Save what the channel shows - but only if the saved post is still exactly as it was when it was read
        (`expected` is its updated_at then), so a change made through the bot meanwhile is never overwritten."""
        async with self.Session() as s:
            res = await s.execute(
                update(Post).where(Post.id == pid, Post.updated_at == expected).values(**fields, updated_at=utcnow())
            )
            await s.commit()
            return bool(res.rowcount)

    # ------------------------------------------------------------ timed deletions
    async def schedule_deletes(self, rows: list) -> None:
        """rows: dicts with channel_id, message_id, delete_at, group_id, created_by."""
        if not rows:
            return
        async with self.Session() as s:
            s.add_all([ScheduledDelete(**r) for r in rows])
            await s.commit()

    async def due_deletes(self, now: datetime, limit: int = 500) -> list:
        async with self.Session() as s:
            q = (
                select(ScheduledDelete)
                .where(ScheduledDelete.status == "pending", ScheduledDelete.delete_at <= now)
                .order_by(ScheduledDelete.delete_at, ScheduledDelete.id)
                .limit(limit)
            )
            return list((await s.execute(q)).scalars())

    async def next_delete_at(self) -> Optional[datetime]:
        async with self.Session() as s:
            q = select(func.min(ScheduledDelete.delete_at)).where(ScheduledDelete.status == "pending")
            return as_utc((await s.execute(q)).scalar_one_or_none())

    async def pending_deletes(self, group_id: Optional[str] = None, status: str = "pending") -> list:
        async with self.Session() as s:
            q = select(ScheduledDelete).where(ScheduledDelete.status == status).order_by(ScheduledDelete.id)
            if group_id:
                q = q.where(ScheduledDelete.group_id == group_id)
            return list((await s.execute(q)).scalars())

    async def finish_deletes(self, ids: list) -> None:
        if not ids:
            return
        async with self.Session() as s:
            for row in (await s.execute(select(ScheduledDelete).where(ScheduledDelete.id.in_(list(ids))))).scalars():
                await s.delete(row)
            await s.commit()

    async def retry_delete(self, sid: int, *, at: datetime, error: str) -> None:
        async with self.Session() as s:
            row = await s.get(ScheduledDelete, sid)
            if row is not None:
                row.attempts += 1
                row.delete_at = at
                row.last_error = error[:160]
                await s.commit()

    async def fail_delete(self, sid: int, error: str) -> None:
        async with self.Session() as s:
            row = await s.get(ScheduledDelete, sid)
            if row is not None:
                row.attempts += 1
                row.status = "failed"
                row.last_error = error[:160]
                await s.commit()

    # ------------------------------------------------------------- replace history
    async def create_batch(self, batch_id: str, old: str, new: str, user_id: int) -> None:
        async with self.Session() as s:
            s.add(Batch(id=batch_id, old_username=old, new_username=new, created_by=user_id))
            await s.commit()

    async def _adopt(self, s, channel_id: int, message_id: int, fields: dict, user_id: int) -> None:
        q = select(Post).where(Post.channel_id == channel_id, Post.message_id == message_id)
        p = (await s.execute(q)).scalar_one_or_none()
        if p is not None:
            for k, v in fields.items():
                setattr(p, k, v)
            p.updated_at = utcnow()
        else:
            s.add(
                Post(
                    channel_id=channel_id,
                    message_id=message_id,
                    status="sent",
                    source="adopted",
                    created_by=user_id,
                    sent_at=utcnow(),
                    **fields,
                )
            )

    async def record_changes(self, batch_id: str, channel_id: int, records: list, user_id: int) -> None:
        """records: [{message_id, old_state, new_state, adopt: {post fields}}]"""
        if not records:
            return
        async with self.Session() as s:
            for r in records:
                s.add(
                    Change(
                        batch_id=batch_id,
                        channel_id=channel_id,
                        message_id=r["message_id"],
                        old_state=r["old_state"],
                        new_state=r["new_state"],
                    )
                )
                await self._adopt(s, channel_id, r["message_id"], r["adopt"], user_id)
            await s.commit()

    async def last_batch(self) -> Optional[Batch]:
        async with self.Session() as s:
            q = select(Batch).where(Batch.undone.is_(False)).order_by(desc(Batch.created_at)).limit(1)
            return (await s.execute(q)).scalar_one_or_none()

    async def get_batch(self, batch_id: str) -> Optional[Batch]:
        async with self.Session() as s:
            return await s.get(Batch, batch_id)

    async def batch_changes(self, batch_id: str) -> list:
        async with self.Session() as s:
            q = select(Change).where(Change.batch_id == batch_id, Change.reverted.is_(False)).order_by(Change.id)
            return list((await s.execute(q)).scalars())

    async def count_batch_changes(self, batch_id: str) -> int:
        async with self.Session() as s:
            q = select(func.count()).select_from(Change).where(Change.batch_id == batch_id, Change.reverted.is_(False))
            return (await s.execute(q)).scalar_one()

    async def finish_undo(self, batch_id: str, change_ids: list, restored: list, all_done: bool) -> None:
        """restored: [(channel_id, message_id, state)] -> refresh the saved post copies."""
        async with self.Session() as s:
            for cid in change_ids:
                c = await s.get(Change, cid)
                if c is not None:
                    c.reverted = True
            for channel_id, message_id, state in restored:
                q = select(Post).where(Post.channel_id == channel_id, Post.message_id == message_id)
                p = (await s.execute(q)).scalar_one_or_none()
                if p is not None:
                    if not state.get("rich"):  # a rich message's text was never changed: it stays as saved
                        p.text = state.get("text", "")
                        p.entities = state.get("entities", [])
                    p.buttons = state.get("buttons", [])
                    p.updated_at = utcnow()
            if all_done:
                b = await s.get(Batch, batch_id)
                if b is not None:
                    b.undone = True
            await s.commit()


    # ------------------------------------------------------------------ /repost
    OPEN_STATES = ("copying", "stopped", "incomplete", "copied")

    async def create_migration(self, mid: str, channel_id: int, *, old, new, include_typed: bool, include_posts: bool,
                               first_id: int, last_id: int, partial: bool, user_id: int) -> None:
        async with self.Session() as s:
            s.add(Migration(id=mid, channel_id=channel_id, old_username=old, new_username=new,
                            include_typed=include_typed, include_posts=include_posts, first_id=first_id,
                            last_id=last_id, partial=partial, created_by=user_id))
            await s.commit()

    async def get_migration(self, mid: str) -> Optional[Migration]:
        async with self.Session() as s:
            return await s.get(Migration, mid)

    async def open_migration(self, channel_id: Optional[int] = None) -> Optional[Migration]:
        """The newest repost that is not finished yet (copies exist, old posts not deleted / copies not removed)."""
        async with self.Session() as s:
            q = select(Migration).where(Migration.status.in_(self.OPEN_STATES))
            if channel_id is not None:
                q = q.where(Migration.channel_id == channel_id)
            q = q.order_by(desc(Migration.created_at)).limit(1)
            return (await s.execute(q)).scalar_one_or_none()

    async def set_migration_status(self, mid: str, status: str) -> None:
        async with self.Session() as s:
            m = await s.get(Migration, mid)
            if m is not None:
                m.status = status
                await s.commit()

    async def migration_done_old_ids(self, mid: str) -> set:
        async with self.Session() as s:
            q = select(MigrationItem.old_id).where(MigrationItem.migration_id == mid)
            return set((await s.execute(q)).scalars())

    async def record_repost(self, mid: str, channel_id: int, rows: list, user_id: int) -> None:
        """rows: [{old_id, new_id, post: {Post fields}}] - saves the mapping and registers the new posts."""
        if not rows:
            return
        async with self.Session() as s:
            await self._drop_slots(s, channel_id, [r["new_id"] for r in rows])
            for r in rows:
                s.add(MigrationItem(migration_id=mid, old_id=r["old_id"], new_id=r["new_id"]))
                s.add(Post(channel_id=channel_id, message_id=r["new_id"], status="sent", source="bot",
                           created_by=user_id, sent_at=utcnow(), **r["post"]))
            await s.commit()

    async def migration_pairs(
        self, mid: Optional[str] = None, channel_id: Optional[int] = None, alive_only: bool = True
    ) -> list:
        """[(old id, new id)] of the copies of one repost (`mid`) or of every repost of a channel, oldest repost first.
        alive_only: leave out copies that were removed again (undo)."""
        async with self.Session() as s:
            q = select(MigrationItem.old_id, MigrationItem.new_id).join(Migration, Migration.id == MigrationItem.migration_id)
            if mid is not None:
                q = q.where(MigrationItem.migration_id == mid)
            if channel_id is not None:
                q = q.where(Migration.channel_id == channel_id)
            if alive_only:
                q = q.where(MigrationItem.new_deleted.is_(False))
            q = q.order_by(Migration.created_at, MigrationItem.id)
            return [(a, b) for a, b in (await s.execute(q)).all()]

    async def update_post_by_message(self, channel_id: int, message_id: int, **fields: Any) -> Optional[Post]:
        """Change the saved copy of the post that sits at `message_id` (None if the bot does not know that post)."""
        async with self.Session() as s:
            q = select(Post).where(Post.channel_id == channel_id, Post.message_id == message_id)
            p = (await s.execute(q)).scalar_one_or_none()
            if p is None:
                return None
            for k, v in fields.items():
                setattr(p, k, v)
            p.updated_at = utcnow()
            await s.commit()
            return p

    async def migration_pending_ids(self, mid: str, which: str, limit: int = 100) -> list:
        col, flag = (MigrationItem.old_id, MigrationItem.old_deleted) if which == "old" else (MigrationItem.new_id, MigrationItem.new_deleted)
        async with self.Session() as s:
            q = select(col).where(MigrationItem.migration_id == mid, flag.is_(False)).order_by(col).limit(limit)
            return list((await s.execute(q)).scalars())

    async def mark_migration_deleted(self, mid: str, channel_id: int, which: str, ids: list) -> None:
        """The messages are gone from Telegram: flag them and drop their saved post copies."""
        if not ids:
            return
        col, flag = ("old_id", "old_deleted") if which == "old" else ("new_id", "new_deleted")
        async with self.Session() as s:
            q = select(MigrationItem).where(MigrationItem.migration_id == mid, getattr(MigrationItem, col).in_(ids))
            for it in (await s.execute(q)).scalars():
                setattr(it, flag, True)
            pq = select(Post).where(Post.channel_id == channel_id, Post.message_id.in_(ids))
            for p in (await s.execute(pq)).scalars():
                await s.delete(p)
            await s.commit()

    async def migration_counts(self, mid: str) -> dict:
        async with self.Session() as s:
            total = (await s.execute(select(func.count()).select_from(MigrationItem).where(MigrationItem.migration_id == mid))).scalar_one()
            old_left = (await s.execute(select(func.count()).select_from(MigrationItem).where(
                MigrationItem.migration_id == mid, MigrationItem.old_deleted.is_(False)))).scalar_one()
            new_left = (await s.execute(select(func.count()).select_from(MigrationItem).where(
                MigrationItem.migration_id == mid, MigrationItem.new_deleted.is_(False)))).scalar_one()
            lo = (await s.execute(select(func.min(MigrationItem.new_id)).where(MigrationItem.migration_id == mid))).scalar_one()
            hi = (await s.execute(select(func.max(MigrationItem.new_id)).where(MigrationItem.migration_id == mid))).scalar_one()
            return {"copied": total, "old_left": old_left, "new_left": new_left, "first_new": lo, "last_new": hi}

    # ------------------------------------------------------------------- /shift
    SHIFT_OPEN = ("copying", "stopped", "incomplete")

    async def create_shift(self, sid: str, *, src, dst_channel_id: int, first_id: int, last_id: int, user_id: int) -> None:
        """`src`: any object with id, access_hash, title and username (the source channel need not be registered)."""
        async with self.Session() as s:
            s.add(Shift(id=sid, src_channel_id=src.id, src_access_hash=src.access_hash or 0, src_title=src.title or "",
                        src_username=src.username, dst_channel_id=dst_channel_id, first_id=first_id, last_id=last_id,
                        created_by=user_id))
            await s.commit()

    async def get_shift(self, sid: str) -> Optional[Shift]:
        async with self.Session() as s:
            return await s.get(Shift, sid)

    async def open_shift(self, src_id: Optional[int] = None, dst_id: Optional[int] = None) -> Optional[Shift]:
        """The newest shift that is not finished (still copying, stopped, or incomplete)."""
        async with self.Session() as s:
            q = select(Shift).where(Shift.status.in_(self.SHIFT_OPEN))
            if src_id is not None:
                q = q.where(Shift.src_channel_id == src_id)
            if dst_id is not None:
                q = q.where(Shift.dst_channel_id == dst_id)
            return (await s.execute(q.order_by(desc(Shift.created_at)).limit(1))).scalar_one_or_none()

    async def last_finished_shift(self) -> Optional[Shift]:
        """The newest finished shift that still has copies in its destination - so its Undo can be offered again."""
        async with self.Session() as s:
            alive = select(ShiftItem.id).where(ShiftItem.shift_id == Shift.id, ShiftItem.new_deleted.is_(False)).exists()
            q = select(Shift).where(Shift.status == "copied", alive).order_by(desc(Shift.created_at)).limit(1)
            return (await s.execute(q)).scalar_one_or_none()

    async def set_shift_status(self, sid: str, status: str) -> None:
        async with self.Session() as s:
            sh = await s.get(Shift, sid)
            if sh is not None:
                sh.status = status
                await s.commit()

    async def shift_done_ids(self, sid: str) -> set:
        async with self.Session() as s:
            q = select(ShiftItem.old_id).where(ShiftItem.shift_id == sid)
            return set((await s.execute(q)).scalars())

    async def shift_pairs(self, sid: str, alive_only: bool = True) -> list:
        """[(source id, id of the copy)] of one shift, in copy order."""
        async with self.Session() as s:
            q = select(ShiftItem.old_id, ShiftItem.new_id).where(ShiftItem.shift_id == sid)
            if alive_only:
                q = q.where(ShiftItem.new_deleted.is_(False))
            return [(a, b) for a, b in (await s.execute(q.order_by(ShiftItem.id))).all()]

    async def shifted_ids(self, src_id: int, dst_id: int, first: int, last: int) -> set:
        """Source ids in first..last that already have a live copy in `dst_id` from an earlier shift."""
        async with self.Session() as s:
            q = (
                select(ShiftItem.old_id)
                .join(Shift, Shift.id == ShiftItem.shift_id)
                .where(Shift.src_channel_id == src_id, Shift.dst_channel_id == dst_id, ShiftItem.new_deleted.is_(False),
                       ShiftItem.old_id >= first, ShiftItem.old_id <= last)
            )
            return set((await s.execute(q)).scalars())

    async def record_shift(self, sid: str, dst_channel_id: int, rows: list, user_id: int) -> None:
        """rows: [{old_id, new_id, post: {Post fields}}] - saves the mapping and registers the copies in My posts."""
        if not rows:
            return
        async with self.Session() as s:
            await self._drop_slots(s, dst_channel_id, [r["new_id"] for r in rows])
            for r in rows:
                s.add(ShiftItem(shift_id=sid, old_id=r["old_id"], new_id=r["new_id"]))
                s.add(Post(channel_id=dst_channel_id, message_id=r["new_id"], status="sent", source="bot",
                           created_by=user_id, sent_at=utcnow(), **r["post"]))
            await s.commit()

    async def shift_counts(self, sid: str) -> dict:
        async with self.Session() as s:
            base = ShiftItem.shift_id == sid
            total = (await s.execute(select(func.count()).select_from(ShiftItem).where(base))).scalar_one()
            left = (await s.execute(select(func.count()).select_from(ShiftItem).where(base, ShiftItem.new_deleted.is_(False)))).scalar_one()
            lo = (await s.execute(select(func.min(ShiftItem.new_id)).where(base))).scalar_one()
            hi = (await s.execute(select(func.max(ShiftItem.new_id)).where(base))).scalar_one()
            return {"copied": total, "new_left": left, "first_new": lo, "last_new": hi}

    async def shift_pending_new(self, sid: str, limit: int = 100) -> list:
        async with self.Session() as s:
            q = select(ShiftItem.new_id).where(ShiftItem.shift_id == sid, ShiftItem.new_deleted.is_(False)).order_by(ShiftItem.new_id).limit(limit)
            return list((await s.execute(q)).scalars())

    async def mark_shift_deleted(self, sid: str, dst_channel_id: int, ids: list) -> None:
        """The copies are gone from Telegram: flag them and drop their saved posts."""
        if not ids:
            return
        async with self.Session() as s:
            q = select(ShiftItem).where(ShiftItem.shift_id == sid, ShiftItem.new_id.in_(ids))
            for it in (await s.execute(q)).scalars():
                it.new_deleted = True
            pq = select(Post).where(Post.channel_id == dst_channel_id, Post.message_id.in_(ids))
            for p in (await s.execute(pq)).scalars():
                await s.delete(p)
            await s.commit()

    # -------------------------------------------------------------------- export
    async def export_all(self) -> dict:
        def clean(d: dict) -> dict:
            return {k: (v.isoformat() if isinstance(v, datetime) else v) for k, v in d.items()}

        async with self.Session() as s:
            chans = list((await s.execute(select(Channel))).scalars())
            posts = list((await s.execute(select(Post).order_by(Post.id))).scalars())
            marks = (await s.execute(select(IgnoredPost.channel_id, IgnoredPost.message_id).order_by(IgnoredPost.id))).all()
            return {
                "exported_at": utcnow().isoformat(),
                "channels": [
                    clean({c.name: getattr(ch, c.name) for c in ch.__table__.columns}) for ch in chans
                ],
                "posts": [clean(p.as_dict()) for p in posts],
                "ignored": [{"channel_id": int(c), "message_id": int(m)} for c, m in marks],  # posts forgotten on purpose
            }