diff --git a/.env.example b/.env.example index 21808ad..abbd68e 100644 --- a/.env.example +++ b/.env.example @@ -19,3 +19,10 @@ CHAPTERWISE_ADMIN_PASSWORD=change-me CHAPTERWISE_PUBLIC_URL=http://localhost:8100 # Seconds between automatic scans of the books folder CHAPTERWISE_SCAN_INTERVAL=600 + +# --- Telegram (bot) ----------------------------------------------------------- +TELEGRAM_BOT_TOKEN= +# The only chat the bot answers. Leave empty, send /start to the bot, and it replies with your ID. +TELEGRAM_CHAT_ID= +CHAPTERWISE_DELIVERY_TIME=07:30 +CHAPTERWISE_TIMEZONE=Europe/Rome diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 86c695d..1ed5890 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -98,7 +98,7 @@ jobs: cancel-in-progress: false env: REMOTE: root@${{ secrets.DEPLOY_HOST }} - REMOTE_DIR: /root/chapterwise + REMOTE_DIR: ${{ vars.DEPLOY_DIR || '/root/chapterwise' }} TAG: sha-${{ github.sha }} steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 @@ -161,5 +161,5 @@ jobs: sleep 5 done echo "::error::Health check failed. Recent logs:" - ssh "$REMOTE" "cd $REMOTE_DIR && ./cw.sh ps && ./cw.sh logs --tail 80 migrate web worker" || true + ssh "$REMOTE" "cd $REMOTE_DIR && ./cw.sh ps && ./cw.sh logs --tail 80 migrate web worker bot" || true exit 1 diff --git a/README.md b/README.md index 3aeaf32..09b8a1c 100644 --- a/README.md +++ b/README.md @@ -7,8 +7,9 @@ chapterwise splits a book into sections using its own table of contents, summari same request is never paid for twice. It's built to run next to a self-hosted library such as [BookOrbit](https://github.com/bookorbit/bookorbit), reading the same books folder read-only. -**Status:** Phase 2. A command-line tool plus a self-hosted web app (library, chapters, summaries) -with a background worker, deployed by CI/CD. Telegram delivery comes next (see [Roadmap](#roadmap)). +**Status:** Phase 3. A command-line tool, a self-hosted web app (library, chapters, summaries) with a +background worker, and a personal Telegram bot that sends one summarized section every morning. +Deployed by CI/CD (see [Roadmap](#roadmap)). ``` books folder (read-only) ──► worker: scan · parse · summarize ──► DeepSeek API @@ -16,6 +17,8 @@ books folder (read-only) ──► worker: scan · parse · summarize ──► Postgres (books, sections, job queue, LLM cache) │ browser ──► Caddy ──► web: login · library · book · section + │ + Telegram ◄──── bot: daily delivery · /books · /reading · /next ``` ## Supported formats @@ -69,6 +72,7 @@ Three processes from one image or install, sharing one Postgres database: | `chapterwise migrate` | Create or upgrade the schema (Alembic) | | `chapterwise serve` | Web UI on port 8100 | | `chapterwise worker` | Scans the books folder every `CHAPTERWISE_SCAN_INTERVAL` seconds, parses new or changed files, runs summaries | +| `chapterwise bot` | Telegram bot (long polling, so no open port): daily delivery and commands | - **Job queue:** a Postgres table claimed with `FOR UPDATE SKIP LOCKED`, so there's no Redis to run. - **Retries:** failed jobs retry with exponential backoff, up to 3 attempts. Jobs interrupted by a @@ -87,6 +91,26 @@ chapterwise migrate chapterwise serve & chapterwise worker ``` +## Daily reading on Telegram + +Pick a book with `/books` (or **Read daily on Telegram** on its web page). Every morning at +`CHAPTERWISE_DELIVERY_TIME` (07:30 Europe/Rome by default) the bot sends the next section's summary, +with buttons to get the **Next** one right away, **Pause**, or open it on the web. + +- **Summaries are prepared ahead:** the next 2 sections are queued for the worker, so the morning + message is ready on time. +- **Missed deliveries catch up:** if the bot was down at 07:30, it sends that day's section on its + first check after coming back. Each section is sent once. +- **Next** sends another section now, and tomorrow's delivery still comes. If that summary isn't + ready yet, it arrives as soon as it is. +- **Commands:** `/books [word]` picks a book, `/reading` shows progress with Pause, Resume and Stop, + and `/next` sends the next section now. +- **Private:** only `TELEGRAM_CHAT_ID` is answered. Until it's set, `/start` replies with your chat + ID to put there. +- **Formatting:** summaries use Telegram's HTML subset, so lists become bullet lines and long + summaries are cut at a paragraph boundary. If Telegram ever rejects the formatting, the message + is resent as plain text. + ## Deployment Pushes to `main` run the tests, build the image to `ghcr.io/aminghuf/chapterwise:sha-`, then @@ -97,13 +121,15 @@ deploy over SSH to the VPS. The deploy writes the image tag, backs up the databa 1. **DNS:** `A read → `. 2. **Server:** copy `deploy/server-bootstrap.sh` to the VPS and run `bash server-bootstrap.sh`. It - creates `/root/chapterwise/.env` with generated secrets and asks for the web password and API key. + creates `/root/chapterwise/.env` with generated secrets and asks for the web password, the API key + and the Telegram bot token and chat ID. Re-running it only adds what's missing. 3. **Caddy:** add `deploy/read.aminghuf.com.caddy` to the Caddyfile and reload Caddy. 4. **GitHub secrets** (Settings → Secrets and variables → Actions): - `DEPLOY_SSH_KEY`: a private key allowed to SSH in as root - `DEPLOY_HOST`: the server's IP - `DEPLOY_KNOWN_HOSTS`: the server's host-key line, from `ssh-keyscan -t ed25519 ` checked against the server + - Optional variable `DEPLOY_DIR`: the server folder to deploy to (default `/root/chapterwise`) **On the server:** `cd /root/chapterwise && ./cw.sh ps | logs -f worker | backup`. @@ -124,6 +150,10 @@ Set these in `.env` or the environment: | `CHAPTERWISE_ADMIN_PASSWORD` | | The web UI password | | `CHAPTERWISE_PUBLIC_URL` | `http://localhost:8100` | With `https://`, cookies are marked Secure | | `CHAPTERWISE_SCAN_INTERVAL` | `600` | Seconds between library scans | +| `TELEGRAM_BOT_TOKEN` | | From @BotFather; the bot idles without it | +| `TELEGRAM_CHAT_ID` | | The only chat the bot answers | +| `CHAPTERWISE_DELIVERY_TIME` | `07:30` | Daily delivery, local time | +| `CHAPTERWISE_TIMEZONE` | `Europe/Rome` | Time zone for the delivery time | ## Development @@ -139,7 +169,7 @@ ruff check . && ruff format . - [x] **Phase 1:** parsers, chunking, cached map-reduce summaries, CLI - [x] **Phase 2:** Postgres, background worker, books-folder sync, web UI at read.aminghuf.com, CI/CD to the VPS -- [ ] **Phase 3:** personal Telegram bot with scheduled delivery +- [x] **Phase 3:** personal Telegram bot with scheduled delivery - [ ] **Phase 4:** whole-book summaries, micro-doses (original text plus recall questions), Markdown playbooks as a source ## Notes diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 18bb4ac..6ecf93f 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -5,6 +5,7 @@ name: chapterwise x-app: &app image: ${IMAGE:?set by image.env} restart: unless-stopped + init: true # forwards stop signals and reaps child processes env_file: .env environment: DATABASE_URL: postgresql+psycopg://chapterwise:${POSTGRES_PASSWORD:?}@db:5432/chapterwise @@ -76,3 +77,12 @@ services: depends_on: migrate: condition: service_completed_successfully + + bot: + <<: *app + command: ["chapterwise", "bot"] + healthcheck: + disable: true + depends_on: + migrate: + condition: service_completed_successfully diff --git a/deploy/server-bootstrap.sh b/deploy/server-bootstrap.sh index bdbe664..8eb5def 100755 --- a/deploy/server-bootstrap.sh +++ b/deploy/server-bootstrap.sh @@ -18,6 +18,8 @@ set_if_missing CHAPTERWISE_PUBLIC_URL https://read.aminghuf.com set_if_missing BOOKS_HOST_PATH /root/bookorbit/books set_if_missing CHAPTERWISE_MODEL deepseek-flash set_if_missing TZ Europe/Rome +set_if_missing CHAPTERWISE_TIMEZONE Europe/Rome +set_if_missing CHAPTERWISE_DELIVERY_TIME 07:30 set_if_missing POSTGRES_PASSWORD "$(openssl rand -hex 24)" set_if_missing CHAPTERWISE_SECRET_KEY "$(openssl rand -hex 32)" @@ -31,5 +33,14 @@ if ! grep -q '^DEEPSEEK_API_KEY=' "$ENV"; then echo "DEEPSEEK_API_KEY=$key" >> "$ENV" fi +if ! grep -q '^TELEGRAM_BOT_TOKEN=' "$ENV"; then + read -rsp "Telegram bot token from @BotFather (Enter to add later): " token; echo + echo "TELEGRAM_BOT_TOKEN=$token" >> "$ENV" +fi +if ! grep -q '^TELEGRAM_CHAT_ID=' "$ENV"; then + read -rp "Your Telegram chat ID (Enter if unknown: send /start to the bot and it will tell you): " chat + echo "TELEGRAM_CHAT_ID=$chat" >> "$ENV" +fi + [ -d /root/bookorbit/books ] || echo "Warning: /root/bookorbit/books doesn't exist; set BOOKS_HOST_PATH in $ENV" echo "Ready: $ENV (edit it any time, then run ./cw.sh up -d)" diff --git a/pyproject.toml b/pyproject.toml index 118c1c4..74ab445 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "chapterwise" -version = "0.2.0" +version = "0.3.0" description = "Summarize your books chapter by chapter with an LLM and read them in micro-doses." readme = "README.md" license = "MIT" @@ -24,8 +24,10 @@ dependencies = [ "pypdf>=5.0", "python-dotenv>=1.0", "python-multipart>=0.0.9", + "python-telegram-bot[job-queue]>=21.0", "sqlalchemy>=2.0", "typer>=0.12", + "tzdata>=2024.1", # time zones in slim containers "uvicorn>=0.30", ] diff --git a/src/chapterwise/__init__.py b/src/chapterwise/__init__.py index 1e00ad4..d87904a 100644 --- a/src/chapterwise/__init__.py +++ b/src/chapterwise/__init__.py @@ -1,3 +1,3 @@ """Summarize books chapter by chapter with an LLM.""" -__version__ = "0.2.0" +__version__ = "0.3.0" diff --git a/src/chapterwise/bot.py b/src/chapterwise/bot.py new file mode 100644 index 0000000..1a2a409 --- /dev/null +++ b/src/chapterwise/bot.py @@ -0,0 +1,299 @@ +"""The personal Telegram bot: pick books, receive one summarized section a day, tap Next for more. + +It uses long polling, so the server needs no open port or webhook. Only TELEGRAM_CHAT_ID is served; +until that is set, /start replies with the chat ID to put there. +""" + +from __future__ import annotations + +import asyncio +import logging +from html import escape + +from sqlalchemy import func, select +from sqlalchemy.orm import sessionmaker +from telegram import ( + BotCommand, + InlineKeyboardButton, + InlineKeyboardMarkup, + LinkPreviewOptions, + Update, +) +from telegram.constants import ParseMode +from telegram.error import BadRequest +from telegram.ext import Application, CallbackQueryHandler, CommandHandler, ContextTypes, filters + +from chapterwise import delivery +from chapterwise.config import Settings +from chapterwise.db import DocumentRow, SectionRow +from chapterwise.telegram_format import to_plain_text + +log = logging.getLogger(__name__) + +CHECK_INTERVAL_SECONDS = 60 +BOOKS_PER_PAGE = 10 +COMMANDS = [ + BotCommand("books", "Pick a book to read daily (add a word to search)"), + BotCommand("reading", "Books you're reading, with progress"), + BotCommand("next", "Get the next section now"), + BotCommand("help", "How this works"), +] + + +def keyboard(buttons: list[list[delivery.Button]]) -> InlineKeyboardMarkup | None: + rows = [ + [ + InlineKeyboardButton(label, url=target[4:]) + if target.startswith("url:") + else InlineKeyboardButton(label, callback_data=target[3:]) + for label, target in row + ] + for row in buttons + if row + ] + return InlineKeyboardMarkup(rows) if rows else None + + +class TelegramSender: + """delivery.Sender over the Bot API, with a plain-text fallback if Telegram rejects the HTML.""" + + def __init__(self, bot, chat_id: int): + self.bot = bot + self.chat_id = chat_id + + async def send(self, html: str, buttons: list[list[delivery.Button]]) -> int | None: + markup = keyboard(buttons) + try: + message = await self.bot.send_message( + self.chat_id, + html, + parse_mode=ParseMode.HTML, + reply_markup=markup, + link_preview_options=LinkPreviewOptions(is_disabled=True), + ) + except BadRequest as exc: + log.warning("Telegram rejected the formatted message (%s); sending plain text", exc) + message = await self.bot.send_message(self.chat_id, to_plain_text(html), reply_markup=markup) + return message.message_id + + async def notify(self, text: str) -> None: + await self.bot.send_message(self.chat_id, text) + + +class ChapterwiseBot: + def __init__(self, settings: Settings, sessions: sessionmaker): + settings.require("telegram_token", "database_url") + self.settings = settings + self.sessions = sessions + self.schedule = delivery.Schedule.from_settings(settings) + + # --- wiring ------------------------------------------------------------------------------------------- + + def build(self, base_url: str | None = None) -> Application: + builder = Application.builder().token(self.settings.telegram_token).post_init(self._post_init) + if base_url: # tests point this at a fake Bot API + builder = builder.base_url(base_url) + app = builder.build() + + owner = ( + filters.Chat(chat_id=self.settings.telegram_chat_id) if self.settings.telegram_chat_id else None + ) + app.add_handler(CommandHandler(["start", "help"], self.start)) + if owner is not None: + app.add_handler(CommandHandler("books", self.books, filters=owner)) + app.add_handler(CommandHandler("reading", self.reading, filters=owner)) + app.add_handler(CommandHandler("next", self.next_command, filters=owner)) + app.add_handler(CallbackQueryHandler(self.button)) + app.job_queue.run_repeating( + self.check, interval=CHECK_INTERVAL_SECONDS, first=5, name="deliveries" + ) + return app + + async def _post_init(self, app: Application) -> None: + await app.bot.set_my_commands(COMMANDS) + if self.settings.telegram_chat_id: + log.info( + "Bot ready for chat %s; daily delivery at %s %s", + self.settings.telegram_chat_id, + self.schedule.label(), + self.settings.timezone, + ) + else: + log.warning("TELEGRAM_CHAT_ID is not set: send /start to the bot to get your chat ID") + + def is_owner(self, update: Update) -> bool: + chat = update.effective_chat + return chat is not None and chat.id == self.settings.telegram_chat_id + + async def db(self, fn, *args): + """Run a blocking database function off the event loop.""" + + def run(): + with self.sessions() as session: + return fn(session, *args) + + return await asyncio.to_thread(run) + + # --- deliveries --------------------------------------------------------------------------------------- + + async def check(self, context: ContextTypes.DEFAULT_TYPE) -> None: + sender = TelegramSender(context.bot, self.settings.telegram_chat_id) + try: + await delivery.deliver_due(self.sessions, self.settings, sender) + except Exception: + log.exception("Delivery check failed; retrying next minute") + + # --- commands ----------------------------------------------------------------------------------------- + + async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + chat = update.effective_chat + if not self.settings.telegram_chat_id: + await update.effective_message.reply_text( + f"Hi! Your chat ID is {chat.id}.\n\nAdd TELEGRAM_CHAT_ID={chat.id} to the server's .env " + "and restart the bot. Until then I won't answer anyone." + ) + return + if not self.is_owner(update): + return # stay silent for everyone else + await update.effective_message.reply_text( + "I send you one summarized section of a book every morning at " + f"{self.schedule.label()}.\n\n" + "/books: pick a book (/books word to search)\n" + "/reading: what you're reading, pause or stop\n" + "/next: the next section right now\n\n" + "Under each section: Next ▶ for more, Pause to take a break." + ) + + async def books(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + query = " ".join(context.args or []).strip() + + def find(session): + stmt = ( + select(DocumentRow.id, DocumentRow.title, DocumentRow.author, func.count(SectionRow.id)) + .join(SectionRow, SectionRow.document_id == DocumentRow.id) + .where(DocumentRow.status == "ready") + .group_by(DocumentRow.id) + .order_by(func.lower(DocumentRow.title)) + .limit(BOOKS_PER_PAGE + 1) + ) + if query: + pattern = f"%{query}%" + stmt = stmt.where(DocumentRow.title.ilike(pattern) | DocumentRow.author.ilike(pattern)) + return session.execute(stmt).all() + + rows = await self.db(find) + if not rows: + await update.effective_message.reply_text( + "No books match." if query else "No books are ready yet. Add some to your library first." + ) + return + buttons = [ + [InlineKeyboardButton(f"📖 {title[:48]} ({count})", callback_data=f"sub:{doc_id}")] + for doc_id, title, _author, count in rows[:BOOKS_PER_PAGE] + ] + more = "\n\nMore books exist: narrow it down with /books word." if len(rows) > BOOKS_PER_PAGE else "" + await update.effective_message.reply_text( + f"Tap a book to read it daily (number of sections in brackets).{more}", + reply_markup=InlineKeyboardMarkup(buttons), + ) + + async def reading(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + subs = await self.db( + lambda s: [(sub, total, sub.document.title) for sub, total in delivery.open_subscriptions(s)] + ) + if not subs: + await update.effective_message.reply_text("You're not reading anything. Pick a book with /books.") + return + for sub, total, title in subs: + state = " · paused" if sub.status == "paused" else "" + toggle = ( + ("Resume", f"resume:{sub.id}") if sub.status == "paused" else ("Pause", f"pause:{sub.id}") + ) + await update.effective_message.reply_text( + f"{escape(title, quote=False)}\nNext: section {sub.position + 1} of {total}{state}", + parse_mode=ParseMode.HTML, + reply_markup=InlineKeyboardMarkup( + [ + [ + InlineKeyboardButton("Next ▶", callback_data=f"next:{sub.id}"), + InlineKeyboardButton(toggle[0], callback_data=toggle[1]), + InlineKeyboardButton("Stop", callback_data=f"stop:{sub.id}"), + ] + ] + ), + ) + + async def next_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + subs = await self.db( + lambda s: [(sub.id, sub.document.title) for sub, _ in delivery.open_subscriptions(s)] + ) + if not subs: + await update.effective_message.reply_text("You're not reading anything. Pick a book with /books.") + elif len(subs) == 1: + await self.send_next(subs[0][0], context) + else: + await update.effective_message.reply_text( + "Which book?", + reply_markup=InlineKeyboardMarkup( + [ + [InlineKeyboardButton(title[:60], callback_data=f"next:{sub_id}")] + for sub_id, title in subs + ] + ), + ) + + async def send_next(self, subscription_id: int, context: ContextTypes.DEFAULT_TYPE) -> str: + await self.db(delivery.request_next, subscription_id) + sent = await delivery.deliver_due( + self.sessions, self.settings, TelegramSender(context.bot, self.settings.telegram_chat_id) + ) + if sent: + return "Here you go" + await context.bot.send_message( + self.settings.telegram_chat_id, + "Summarizing the next section… I'll send it as soon as it's ready.", + ) + return "Preparing…" + + # --- buttons ------------------------------------------------------------------------------------------ + + async def button(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + query = update.callback_query + if not self.is_owner(update): + await query.answer() + return + action, _, raw_id = (query.data or "").partition(":") + try: + target = int(raw_id) + except ValueError: + await query.answer("Unknown button") + return + + try: + if action == "sub": + sub = await self.db(delivery.subscribe, target) + title = await self.db(lambda s: s.get(DocumentRow, sub.document_id).title) + await query.answer("Subscribed") + await query.edit_message_text( + f"📚 {title}\n\nYou'll get one section every morning at {self.schedule.label()}. " + "Tap Next to start now.", + reply_markup=InlineKeyboardMarkup( + [[InlineKeyboardButton("Next ▶", callback_data=f"next:{sub.id}")]] + ), + ) + elif action == "next": + await query.answer(await self.send_next(target, context)) + elif action in ("pause", "resume", "stop"): + status = {"pause": "paused", "resume": "active", "stop": "stopped"}[action] + await self.db(delivery.set_status, target, status) + await query.answer({"paused": "Paused", "active": "Resumed", "stopped": "Stopped"}[status]) + if status != "active": + await query.edit_message_reply_markup(reply_markup=None) + else: + await query.answer("Unknown button") + except delivery.SubscriptionError as exc: + await query.answer(str(exc), show_alert=True) + + +def run(settings: Settings, sessions: sessionmaker) -> None: + ChapterwiseBot(settings, sessions).build().run_polling(allowed_updates=Update.ALL_TYPES) diff --git a/src/chapterwise/cli.py b/src/chapterwise/cli.py index 32dbc81..f615661 100644 --- a/src/chapterwise/cli.py +++ b/src/chapterwise/cli.py @@ -1,8 +1,9 @@ -"""Command line: inspect and summarize books, and run the service (migrate, serve, worker).""" +"""Command line: inspect and summarize books, and run the service (migrate, serve, worker, bot).""" from __future__ import annotations import logging +import signal from pathlib import Path from typing import Annotated @@ -200,6 +201,26 @@ def worker() -> None: Worker(settings, make_session_factory(make_engine(settings.database_url))).run_forever() +@app.command() +def bot() -> None: + """Run the Telegram bot: daily deliveries, /books, /reading, /next.""" + from chapterwise.bot import run + from chapterwise.db import make_engine, make_session_factory + + settings = Settings.from_env() + settings.require("database_url") + _logging() + if not settings.telegram_token: + # Idle instead of crash-looping in Docker until a token is configured. + logging.getLogger(__name__).warning("TELEGRAM_BOT_TOKEN is not set; the bot is idle") + stop = {signal.SIGTERM, signal.SIGINT} + signal.pthread_sigmask(signal.SIG_BLOCK, stop) + signal.sigwait(stop) + return + logging.getLogger("httpx").setLevel(logging.WARNING) # don't log every poll (URLs contain the token) + run(settings, make_session_factory(make_engine(settings.database_url))) + + class _NoLLM: """Stand-in for --dry-run: knows the model name for cache keys, refuses to be called.""" diff --git a/src/chapterwise/config.py b/src/chapterwise/config.py index 4264f25..f2f4423 100644 --- a/src/chapterwise/config.py +++ b/src/chapterwise/config.py @@ -44,6 +44,11 @@ class Settings: admin_password: str | None = None public_url: str = "http://localhost:8100" scan_interval: int = 600 # seconds between library scans + # Telegram delivery + telegram_token: str | None = None + telegram_chat_id: int | None = None # the only chat the bot answers + delivery_time: str = "07:30" # local time of the daily delivery + timezone: str = "Europe/Rome" @property def secure_cookies(self) -> bool: @@ -65,6 +70,10 @@ def from_env(cls) -> Settings: admin_password=env("CHAPTERWISE_ADMIN_PASSWORD") or None, public_url=(env("CHAPTERWISE_PUBLIC_URL") or "http://localhost:8100").rstrip("/"), scan_interval=_int("CHAPTERWISE_SCAN_INTERVAL", 600), + telegram_token=env("TELEGRAM_BOT_TOKEN") or None, + telegram_chat_id=int(chat) if (chat := (env("TELEGRAM_CHAT_ID") or "").strip()) else None, + delivery_time=(env("CHAPTERWISE_DELIVERY_TIME") or "07:30").strip(), + timezone=(env("CHAPTERWISE_TIMEZONE") or env("TZ") or "Europe/Rome").strip(), ) def require(self, *names: str) -> None: @@ -73,6 +82,7 @@ def require(self, *names: str) -> None: "database_url": "DATABASE_URL", "secret_key": "CHAPTERWISE_SECRET_KEY", "admin_password": "CHAPTERWISE_ADMIN_PASSWORD", + "telegram_token": "TELEGRAM_BOT_TOKEN", } missing = [env_names.get(n, n) for n in names if not getattr(self, n)] if missing: diff --git a/src/chapterwise/db.py b/src/chapterwise/db.py index cb0e93f..eeff257 100644 --- a/src/chapterwise/db.py +++ b/src/chapterwise/db.py @@ -2,11 +2,12 @@ from __future__ import annotations -from datetime import datetime +from datetime import date, datetime from typing import Any from sqlalchemy import ( BigInteger, + Boolean, Double, ForeignKey, Index, @@ -21,9 +22,10 @@ from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.engine import Engine from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker -from sqlalchemy.types import DateTime +from sqlalchemy.types import Date, DateTime ACTIVE_JOB_STATUSES = ("queued", "running") +OPEN_SUBSCRIPTION_STATUSES = ("active", "paused") class Base(DeclarativeBase): @@ -116,6 +118,48 @@ class CacheEntry(Base): created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) +class Subscription(Base): + """A book being read on Telegram, one section a day.""" + + __tablename__ = "subscriptions" + __table_args__ = ( + # One open (active or paused) subscription per book. + Index( + "subscriptions_one_open_per_book", + "document_id", + unique=True, + postgresql_where=text("status IN ('active', 'paused')"), + ), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + document_id: Mapped[int] = mapped_column(ForeignKey("documents.id", ondelete="CASCADE"), index=True) + mode: Mapped[str] = mapped_column(String(16), default="summary") # "dose" arrives in Phase 4 + position: Mapped[int] = mapped_column(Integer, default=0) # next section to deliver + status: Mapped[str] = mapped_column(String(16), default="active") # active|paused|finished|stopped + wants_next: Mapped[bool] = mapped_column(Boolean, default=False) # "Next" pressed, waiting for summary + last_delivered_on: Mapped[date | None] = mapped_column(Date) # local date of the last daily delivery + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + document: Mapped[DocumentRow] = relationship() + + +class Delivery(Base): + __tablename__ = "deliveries" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + subscription_id: Mapped[int] = mapped_column( + ForeignKey("subscriptions.id", ondelete="CASCADE"), index=True + ) + position: Mapped[int] = mapped_column(Integer) + kind: Mapped[str] = mapped_column(String(16)) # scheduled | manual + message_id: Mapped[int | None] = mapped_column(BigInteger) + sent_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + def make_engine(database_url: str) -> Engine: return create_engine(database_url, pool_pre_ping=True) diff --git a/src/chapterwise/delivery.py b/src/chapterwise/delivery.py new file mode 100644 index 0000000..70730ba --- /dev/null +++ b/src/chapterwise/delivery.py @@ -0,0 +1,278 @@ +"""Daily reading on Telegram: which section is due, and what happens after it is sent. + +Everything here is plain database logic plus one async loop (`deliver_due`) that talks to a `Sender`. +The Telegram specifics live in bot.py, so this module is tested with a fake sender. +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass +from datetime import date, datetime, time +from html import escape +from typing import Protocol +from zoneinfo import ZoneInfo + +from sqlalchemy import func, select +from sqlalchemy.orm import Session, sessionmaker + +from chapterwise import jobs +from chapterwise.cache import PostgresCache +from chapterwise.config import Settings +from chapterwise.db import OPEN_SUBSCRIPTION_STATUSES, Delivery, DocumentRow, SectionRow, Subscription +from chapterwise.summarize import cache_key +from chapterwise.telegram_format import build_message + +log = logging.getLogger(__name__) + +PREFETCH = 2 # summaries prepared ahead, so the morning message is ready on time + + +class SubscriptionError(Exception): + """A user-facing reason a subscription action can't be done.""" + + +@dataclass(frozen=True) +class Schedule: + at: time + tz: ZoneInfo + + @classmethod + def from_settings(cls, settings: Settings) -> Schedule: + hours, _, minutes = settings.delivery_time.partition(":") + return cls(at=time(int(hours), int(minutes or 0)), tz=ZoneInfo(settings.timezone)) + + def today(self, now: datetime) -> date: + return now.astimezone(self.tz).date() + + def is_past_delivery_time(self, now: datetime) -> bool: + return now.astimezone(self.tz).time() >= self.at + + def label(self) -> str: + return self.at.strftime("%H:%M") + + +@dataclass(frozen=True) +class Item: + subscription_id: int + document_id: int + book_title: str + section_title: str + position: int + total: int + summary: str + kind: str # scheduled | manual + + @property + def is_last(self) -> bool: + return self.position + 1 >= self.total + + +# --- subscriptions --------------------------------------------------------------------------------------- + + +def section_count(session: Session, document_id: int) -> int: + return session.scalar(select(func.count()).where(SectionRow.document_id == document_id)) or 0 + + +def open_subscription(session: Session, document_id: int) -> Subscription | None: + return session.scalar( + select(Subscription).where( + Subscription.document_id == document_id, Subscription.status.in_(OPEN_SUBSCRIPTION_STATUSES) + ) + ) + + +def subscribe(session: Session, document_id: int, start: int = 0) -> Subscription: + """Start reading a book daily (or resume the open subscription for it).""" + doc = session.get(DocumentRow, document_id) + if doc is None or doc.status != "ready": + raise SubscriptionError("That book isn't ready yet.") + total = section_count(session, document_id) + if not 0 <= start < total: + raise SubscriptionError("That book has no such section.") + sub = open_subscription(session, document_id) + if sub is None: + sub = Subscription(document_id=document_id, position=start, status="active", mode="summary") + session.add(sub) + else: + sub.status = "active" + session.flush() + prefetch(session, sub) + session.commit() + return sub + + +def set_status(session: Session, subscription_id: int, status: str) -> Subscription: + if status not in ("active", "paused", "stopped"): + raise ValueError(status) + sub = session.get(Subscription, subscription_id) + if sub is None or sub.status not in OPEN_SUBSCRIPTION_STATUSES: + raise SubscriptionError("That subscription has ended.") + sub.status = status + if status != "active": + sub.wants_next = False + session.commit() + return sub + + +def request_next(session: Session, subscription_id: int) -> Subscription: + """'Next' was tapped: deliver the next section as soon as its summary is ready.""" + sub = session.get(Subscription, subscription_id) + if sub is None or sub.status not in OPEN_SUBSCRIPTION_STATUSES: + raise SubscriptionError("That book is finished or stopped. Start it again with /books.") + sub.status = "active" + sub.wants_next = True + prefetch(session, sub) + session.commit() + return sub + + +def open_subscriptions(session: Session) -> list[tuple[Subscription, int]]: + """Active and paused subscriptions with their section counts, newest first.""" + subs = session.scalars( + select(Subscription) + .where(Subscription.status.in_(OPEN_SUBSCRIPTION_STATUSES)) + .order_by(Subscription.id.desc()) + ).all() + return [(sub, section_count(session, sub.document_id)) for sub in subs] + + +def prefetch(session: Session, sub: Subscription, ahead: int = PREFETCH) -> None: + """Queue summaries for the next sections so they're ready before they're due.""" + for position in range(sub.position, sub.position + ahead + 1): + exists = session.scalar( + select(SectionRow.id).where( + SectionRow.document_id == sub.document_id, SectionRow.position == position + ) + ) + if exists: + jobs.enqueue( + session, + "summarize", + {"document_id": sub.document_id, "position": position}, + dedupe_key=f"summarize:{sub.document_id}:{position}", + ) + + +# --- what's due ------------------------------------------------------------------------------------------ + + +def due_items( + session: Session, cache: PostgresCache, settings: Settings, schedule: Schedule, now: datetime +) -> list[Item]: + """Sections to send now. Summaries that aren't ready yet are queued and picked up on a later check.""" + today = schedule.today(now) + daily_due = schedule.is_past_delivery_time(now) + items: list[Item] = [] + + for sub in session.scalars(select(Subscription).where(Subscription.status == "active")): + scheduled = daily_due and sub.last_delivered_on != today + if not (scheduled or sub.wants_next): + continue + doc = session.get(DocumentRow, sub.document_id) + total = section_count(session, sub.document_id) + section = session.scalar( + select(SectionRow).where( + SectionRow.document_id == sub.document_id, SectionRow.position == sub.position + ) + ) + if doc is None or section is None: # past the end, e.g. after a re-parse shortened the book + sub.status, sub.wants_next = "finished", False + continue + entry = cache.get(cache_key("section", section.text_sha256, doc.title, section.title, settings.model)) + if entry is None: + prefetch(session, sub) + continue + items.append( + Item( + subscription_id=sub.id, + document_id=doc.id, + book_title=doc.title, + section_title=section.title, + position=sub.position, + total=total, + summary=entry["text"], + kind="scheduled" if scheduled else "manual", + ) + ) + session.commit() + return items + + +def record_delivery( + session: Session, item: Item, message_id: int | None, schedule: Schedule, now: datetime +) -> Subscription: + """Move the subscription past the section just sent and prepare the next ones.""" + sub = session.get(Subscription, item.subscription_id) + session.add( + Delivery(subscription_id=sub.id, position=item.position, kind=item.kind, message_id=message_id) + ) + sub.position = item.position + 1 + sub.wants_next = False + if item.kind == "scheduled": + sub.last_delivered_on = schedule.today(now) + if item.is_last: + sub.status = "finished" + else: + prefetch(session, sub) + session.commit() + return sub + + +# --- sending --------------------------------------------------------------------------------------------- + +Button = tuple[str, str] # (label, "cb:" or "url:") + + +class Sender(Protocol): + async def send(self, html: str, buttons: list[list[Button]]) -> int | None: ... + + async def notify(self, text: str) -> None: ... + + +def message_for(item: Item, settings: Settings) -> tuple[str, list[list[Button]]]: + header = ( + f"{escape(item.book_title, quote=False)}\n" + f"{escape(item.section_title, quote=False)} · {item.position + 1}/{item.total}" + ) + html, _truncated = build_message(header, item.summary) + row: list[Button] = [] + if not item.is_last: + row.append(("Next ▶", f"cb:next:{item.subscription_id}")) + row.append(("Pause", f"cb:pause:{item.subscription_id}")) + buttons = [row] + if settings.public_url.startswith("https://"): # Telegram rejects non-public button URLs + url = f"{settings.public_url}/b/{item.document_id}/s/{item.position}" + buttons.append([("Open on the web", f"url:{url}")]) + return html, buttons + + +async def deliver_due( + sessions: sessionmaker, settings: Settings, sender: Sender, now: datetime | None = None +) -> int: + """Send everything that's due. Safe to call often: each section is sent once.""" + schedule = Schedule.from_settings(settings) + cache = PostgresCache(sessions) + now = now or datetime.now(schedule.tz) + + def load() -> list[Item]: + with sessions() as session: + return due_items(session, cache, settings, schedule, now) + + sent = 0 + for item in await asyncio.to_thread(load): + html, buttons = message_for(item, settings) + message_id = await sender.send(html, buttons) + + def save(item: Item = item, message_id: int | None = message_id) -> None: + with sessions() as session: + record_delivery(session, item, message_id, schedule, now) + + await asyncio.to_thread(save) + sent += 1 + log.info("Delivered '%s' %d/%d (%s)", item.book_title, item.position + 1, item.total, item.kind) + if item.is_last: + await sender.notify(f"🎉 You finished {item.book_title}! Pick the next one with /books.") + return sent diff --git a/src/chapterwise/jobs.py b/src/chapterwise/jobs.py index fbf9550..baf1f01 100644 --- a/src/chapterwise/jobs.py +++ b/src/chapterwise/jobs.py @@ -5,13 +5,14 @@ from datetime import UTC, datetime, timedelta from typing import Any -from sqlalchemy import func, select, update +from sqlalchemy import func, select, text, update from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import Session from chapterwise.db import ACTIVE_JOB_STATUSES, Job MAX_ATTEMPTS = 3 +ACTIVE_DEDUPE_PREDICATE = text("status IN ('queued', 'running')") # same as the jobs_active_dedupe index RETRY_BASE_SECONDS = 30 @@ -22,9 +23,9 @@ def enqueue( stmt = ( insert(Job) .values(kind=kind, payload=payload or {}, dedupe_key=dedupe_key, status="queued") - .on_conflict_do_nothing( - index_elements=[Job.dedupe_key], index_where=Job.status.in_(ACTIVE_JOB_STATUSES) - ) + # The predicate must be literal SQL: with bound parameters, Postgres switches to a generic plan + # after a few executions and can no longer match the partial unique index. + .on_conflict_do_nothing(index_elements=[Job.dedupe_key], index_where=ACTIVE_DEDUPE_PREDICATE) .returning(Job.id) ) return session.execute(stmt).first() is not None diff --git a/src/chapterwise/migrations/versions/0002_subscriptions_and_deliveries.py b/src/chapterwise/migrations/versions/0002_subscriptions_and_deliveries.py new file mode 100644 index 0000000..7404a9e --- /dev/null +++ b/src/chapterwise/migrations/versions/0002_subscriptions_and_deliveries.py @@ -0,0 +1,65 @@ +"""subscriptions and deliveries + +Revision ID: 0002 +Revises: 0001 +Create Date: 2026-11-01 00:49:10.354298 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0002" +down_revision: str | None = "0001" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "subscriptions", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("document_id", sa.Integer(), nullable=False), + sa.Column("mode", sa.String(length=16), nullable=False), + sa.Column("position", sa.Integer(), nullable=False), + sa.Column("status", sa.String(length=16), nullable=False), + sa.Column("wants_next", sa.Boolean(), nullable=False), + sa.Column("last_delivered_on", sa.Date(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["document_id"], ["documents.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_subscriptions_document_id"), "subscriptions", ["document_id"], unique=False) + op.create_index( + "subscriptions_one_open_per_book", + "subscriptions", + ["document_id"], + unique=True, + postgresql_where=sa.text("status IN ('active', 'paused')"), + ) + op.create_table( + "deliveries", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("subscription_id", sa.Integer(), nullable=False), + sa.Column("position", sa.Integer(), nullable=False), + sa.Column("kind", sa.String(length=16), nullable=False), + sa.Column("message_id", sa.BigInteger(), nullable=True), + sa.Column("sent_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["subscription_id"], ["subscriptions.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_deliveries_subscription_id"), "deliveries", ["subscription_id"], unique=False) + + +def downgrade() -> None: + op.drop_index(op.f("ix_deliveries_subscription_id"), table_name="deliveries") + op.drop_table("deliveries") + op.drop_index( + "subscriptions_one_open_per_book", + table_name="subscriptions", + postgresql_where=sa.text("status IN ('active', 'paused')"), + ) + op.drop_index(op.f("ix_subscriptions_document_id"), table_name="subscriptions") + op.drop_table("subscriptions") diff --git a/src/chapterwise/telegram_format.py b/src/chapterwise/telegram_format.py new file mode 100644 index 0000000..7b82433 --- /dev/null +++ b/src/chapterwise/telegram_format.py @@ -0,0 +1,107 @@ +"""Turn Markdown summaries into the small HTML subset Telegram accepts. + +Telegram supports only 
  
, no lists or headings, and caps a
+message at 4096 characters. Lists become bullet lines, headings become bold lines, and long summaries
+are cut at a block boundary so no tag is ever left open.
+"""
+
+from __future__ import annotations
+
+from html import escape
+
+from bs4 import BeautifulSoup, NavigableString, Tag
+from markdown_it import MarkdownIt
+
+MESSAGE_BUDGET = 3800  # leaves room under Telegram's 4096-character limit
+
+_markdown = MarkdownIt("commonmark", {"html": False}).enable(["table", "strikethrough"])
+_INLINE = {"strong": "b", "b": "b", "em": "i", "i": "i", "del": "s", "s": "s", "u": "u"}
+
+
+def markdown_to_blocks(markdown: str) -> list[str]:
+    """Telegram-HTML blocks (paragraphs, list items, ...) in reading order."""
+    soup = BeautifulSoup(_markdown.render(markdown), "html.parser")
+    blocks: list[str] = []
+    for node in soup.children:
+        if isinstance(node, Tag):
+            blocks.extend(_block(node))
+        elif str(node).strip():
+            blocks.append(escape(str(node).strip(), quote=False))
+    return [b for b in blocks if b.strip()]
+
+
+def _block(tag: Tag, depth: int = 0) -> list[str]:
+    name = tag.name
+    if name in ("h1", "h2", "h3", "h4", "h5", "h6"):
+        return [f"{_inline(tag)}"]
+    if name in ("ul", "ol"):
+        lines: list[str] = []
+        for number, item in enumerate(tag.find_all("li", recursive=False), start=1):
+            marker = f"{number}." if name == "ol" else "•"
+            nested = [
+                child for child in item.children if isinstance(child, Tag) and child.name in ("ul", "ol")
+            ]
+            for child in nested:
+                child.extract()
+            lines.append(f"{'  ' * depth}{marker} {_inline(item).strip()}")
+            for child in nested:
+                lines.extend(_block(child, depth + 1))
+        return ["\n".join(lines)] if depth == 0 else lines
+    if name == "pre":
+        return [f"
{escape(tag.get_text(), quote=False)}
"]
+    if name == "blockquote":
+        return [f"
{_inline(tag).strip()}
"]
+    if name == "hr":
+        return ["———"]
+    if name == "table":
+        rows = [
+            " | ".join(_inline(cell).strip() for cell in row.find_all(["th", "td"]))
+            for row in tag.find_all("tr")
+        ]
+        return ["\n".join(rows)]
+    return [_inline(tag).strip()]
+
+
+def _inline(node) -> str:
+    out: list[str] = []
+    for child in node.children:
+        if isinstance(child, NavigableString):
+            out.append(escape(str(child), quote=False))
+            continue
+        name = child.name
+        if name in _INLINE:
+            tag = _INLINE[name]
+            out.append(f"<{tag}>{_inline(child)}")
+        elif name == "code":
+            out.append(f"{escape(child.get_text(), quote=False)}")
+        elif name == "a":
+            href = child.get("href", "")
+            text = _inline(child)
+            out.append(
+                f'{text}' if href.startswith(("http://", "https://")) else text
+            )
+        elif name == "br":
+            out.append("\n")
+        elif name == "p":
+            out.append(_inline(child) + " ")
+        else:
+            out.append(_inline(child))
+    return "".join(out).replace("\n ", "\n")
+
+
+def build_message(header: str, markdown: str, budget: int = MESSAGE_BUDGET) -> tuple[str, bool]:
+    """Header (already HTML) + summary, cut at a block boundary to fit. Returns (html, truncated)."""
+    parts = [header]
+    used = len(header)
+    for block in markdown_to_blocks(markdown):
+        if used + len(block) + 2 > budget:
+            parts.append("…")
+            return "\n\n".join(parts), True
+        parts.append(block)
+        used += len(block) + 2
+    return "\n\n".join(parts), False
+
+
+def to_plain_text(html: str) -> str:
+    """Fallback if Telegram ever rejects the HTML: same content without markup."""
+    return BeautifulSoup(html, "html.parser").get_text()
diff --git a/src/chapterwise/templates/book.html b/src/chapterwise/templates/book.html
index edf8162..2685300 100644
--- a/src/chapterwise/templates/book.html
+++ b/src/chapterwise/templates/book.html
@@ -13,6 +13,26 @@ 
{{ doc.title }}

   
This file is no longer in the books folder. Existing summaries are kept.

 {% endif %}
 
+{% if rows and telegram %}
+

+  {% if subscription %}
+    📬 Daily on Telegram at {{ delivery_time }}: next is section {{ subscription.position }}{% if subscription.status == "paused" %} paused{% endif %}
+    {% if subscription.status == "paused" %}
+      

+    {% else %}
+      

+    {% endif %}
+    

+  {% else %}
+    

+      
+      
+    

+    one section every morning at {{ delivery_time }}
+  {% endif %}
+

+{% endif %}
+
 {% if rows %}
 

   {% if missing_count %}
diff --git a/src/chapterwise/web.py b/src/chapterwise/web.py
index 7cfdf09..9eee163 100644
--- a/src/chapterwise/web.py
+++ b/src/chapterwise/web.py
@@ -16,7 +16,7 @@
 from sqlalchemy.orm import Session, sessionmaker
 from starlette.middleware.sessions import SessionMiddleware
 
-from chapterwise import jobs
+from chapterwise import delivery, jobs
 from chapterwise.cache import PostgresCache, cache_digest
 from chapterwise.config import Settings
 from chapterwise.db import (
@@ -254,8 +254,32 @@ def book(request: Request, document_id: int, session: DB, _user: User):
             missing_count=len(missing),
             missing_tokens=sum(s.tokens for s in missing),
             refresh=doc.status == "pending" or bool(active),
+            telegram=bool(settings.telegram_token),
+            subscription=delivery.open_subscription(session, doc.id),
+            delivery_time=settings.delivery_time,
         )
 
+    # --- daily reading on Telegram ------------------------------------------------------------------------
+
+    @app.post("/b/{document_id}/subscribe")
+    def subscribe(document_id: int, session: DB, _user: User, start: Annotated[int, Form()] = 0):
+        try:
+            delivery.subscribe(session, document_id, start)
+        except delivery.SubscriptionError as exc:
+            raise HTTPException(400, str(exc)) from None
+        return RedirectResponse(f"/b/{document_id}", status_code=303)
+
+    @app.post("/subscriptions/{subscription_id}/{action}")
+    def change_subscription(subscription_id: int, action: str, session: DB, _user: User):
+        status = {"pause": "paused", "resume": "active", "stop": "stopped"}.get(action)
+        if status is None:
+            raise HTTPException(404)
+        try:
+            sub = delivery.set_status(session, subscription_id, status)
+        except delivery.SubscriptionError as exc:
+            raise HTTPException(400, str(exc)) from None
+        return RedirectResponse(f"/b/{sub.document_id}", status_code=303)
+
     @app.post("/b/{document_id}/summarize")
     def summarize(document_id: int, session: DB, _user: User, sections: Annotated[str, Form()]):
         if not settings.api_key:
diff --git a/tests/test_bot.py b/tests/test_bot.py
new file mode 100644
index 0000000..430cf3c
--- /dev/null
+++ b/tests/test_bot.py
@@ -0,0 +1,248 @@
+"""The bot end to end: real python-telegram-bot, real HTTP, against a local fake of the Bot API."""
+
+import asyncio
+import json
+import threading
+from dataclasses import replace
+from datetime import datetime
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import parse_qs
+from zoneinfo import ZoneInfo
+
+import pytest
+from sqlalchemy import select
+from telegram import Update
+
+from chapterwise import delivery
+from chapterwise.bot import ChapterwiseBot, TelegramSender
+from chapterwise.db import DocumentRow, Subscription
+from chapterwise.worker import Worker
+from tests.conftest import FakeLLM
+
+OWNER = 4242
+STRANGER = 999
+
+
+class FakeBotAPI:
+    """Records every Bot API call and answers just enough for python-telegram-bot."""
+
+    def __init__(self):
+        self.calls: list[tuple[str, dict]] = []
+        self.reject_html = False
+        api = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def do_POST(self):
+                method = self.path.rsplit("/", 1)[-1]
+                raw = self.rfile.read(int(self.headers.get("Content-Length") or 0)).decode()
+                if "json" in (self.headers.get("Content-Type") or ""):
+                    params = json.loads(raw or "{}")
+                else:
+                    params = {k: v[0] for k, v in parse_qs(raw).items()}
+                api.calls.append((method, params))
+                status, payload = api.answer(method, params)
+                data = json.dumps(payload).encode()
+                self.send_response(status)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(data)))
+                self.end_headers()
+                self.wfile.write(data)
+
+            def log_message(self, *args):
+                pass
+
+        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        threading.Thread(target=self.server.serve_forever, daemon=True).start()
+        self.base_url = f"http://127.0.0.1:{self.server.server_port}/bot"
+
+    def answer(self, method: str, params: dict):
+        if method == "getMe":
+            return 200, {
+                "ok": True,
+                "result": {"id": 1, "is_bot": True, "first_name": "cw", "username": "cw_bot"},
+            }
+        if method == "sendMessage":
+            if self.reject_html and params.get("parse_mode") == "HTML":
+                return 400, {
+                    "ok": False,
+                    "error_code": 400,
+                    "description": "Bad Request: can't parse entities",
+                }
+            message_id = 100 + len(self.sent())
+            chat = {"id": int(params["chat_id"]), "type": "private"}
+            return 200, {
+                "ok": True,
+                "result": {"message_id": message_id, "date": 0, "chat": chat, "text": params["text"]},
+            }
+        if method == "getUpdates":
+            return 200, {"ok": True, "result": []}
+        if method in ("editMessageText", "editMessageReplyMarkup"):
+            return 200, {
+                "ok": True,
+                "result": {"message_id": 7, "date": 0, "chat": {"id": OWNER, "type": "private"}, "text": "x"},
+            }
+        return 200, {"ok": True, "result": True}
+
+    def sent(self) -> list[dict]:
+        return [p for m, p in self.calls if m == "sendMessage"]
+
+    def calls_of(self, method: str) -> list[dict]:
+        return [p for m, p in self.calls if m == method]
+
+
+def command(text: str, chat_id: int = OWNER, update_id: int = 1) -> dict:
+    word = text.split()[0]
+    return {
+        "update_id": update_id,
+        "message": {
+            "message_id": update_id,
+            "date": 0,
+            "chat": {"id": chat_id, "type": "private"},
+            "from": {"id": chat_id, "is_bot": False, "first_name": "Amin"},
+            "text": text,
+            "entities": [{"type": "bot_command", "offset": 0, "length": len(word)}],
+        },
+    }
+
+
+def tap(data: str, chat_id: int = OWNER, update_id: int = 1) -> dict:
+    return {
+        "update_id": update_id,
+        "callback_query": {
+            "id": f"cb{update_id}",
+            "from": {"id": chat_id, "is_bot": False, "first_name": "Amin"},
+            "chat_instance": "x",
+            "data": data,
+            "message": {
+                "message_id": 7,
+                "date": 0,
+                "chat": {"id": chat_id, "type": "private"},
+                "text": "earlier message",
+            },
+        },
+    }
+
+
+@pytest.fixture
+def api(monkeypatch):
+    monkeypatch.setenv("NO_PROXY", "127.0.0.1,localhost")
+    fake = FakeBotAPI()
+    yield fake
+    fake.server.shutdown()
+
+
+@pytest.fixture
+def bot_settings(settings):
+    return replace(settings, telegram_token="123:abc", telegram_chat_id=OWNER)
+
+
+@pytest.fixture
+def worker(sessions, bot_settings):
+    w = Worker(bot_settings, sessions, llm=FakeLLM("fake-model"))
+    w.schedule_scan()
+    while w.run_once():
+        pass
+    return w
+
+
+def run_updates(bot: ChapterwiseBot, api: FakeBotAPI, *updates: dict) -> None:
+    async def go():
+        app = bot.build(base_url=api.base_url)
+        async with app:  # initialize (getMe) without starting polling or the job queue
+            for data in updates:
+                await app.process_update(Update.de_json(data, app.bot))
+
+    asyncio.run(go())
+
+
+def test_unconfigured_bot_tells_you_your_chat_id(sessions, bot_settings, api):
+    bot = ChapterwiseBot(replace(bot_settings, telegram_chat_id=None), sessions)
+    run_updates(bot, api, command("/start", chat_id=STRANGER))
+    assert "Your chat ID is 999" in api.sent()[0]["text"]
+
+
+def test_strangers_are_ignored(sessions, bot_settings, worker, api):
+    bot = ChapterwiseBot(bot_settings, sessions)
+    run_updates(
+        bot, api, command("/start", chat_id=STRANGER), command("/books", chat_id=STRANGER, update_id=2)
+    )
+    assert api.sent() == []
+    run_updates(bot, api, tap("sub:1", chat_id=STRANGER))
+    assert api.sent() == []
+    with sessions() as s:
+        assert s.scalar(select(Subscription)) is None
+
+
+def test_pick_a_book_then_next(sessions, bot_settings, worker, api):
+    bot = ChapterwiseBot(bot_settings, sessions)
+    with sessions() as s:
+        epub = s.scalar(select(DocumentRow.id).where(DocumentRow.format == "epub"))
+
+    run_updates(bot, api, command("/books test"))
+    keyboard = json.loads(api.sent()[0]["reply_markup"])["inline_keyboard"]
+    assert keyboard == [[{"text": "📖 The Test Book (3)", "callback_data": f"sub:{epub}"}]]
+
+    run_updates(bot, api, tap(f"sub:{epub}"))
+    edited = api.calls_of("editMessageText")[0]
+    assert "every morning at 07:30" in edited["text"]
+    with sessions() as s:
+        sub = s.scalar(select(Subscription))
+        assert (sub.document_id, sub.status) == (epub, "active")
+
+    # Next before the summary exists: the bot says it's preparing...
+    run_updates(bot, api, tap(f"next:{sub.id}", update_id=2))
+    assert "I'll send it as soon as it's ready" in api.sent()[-1]["text"]
+    # ...the worker summarizes, and the next check sends it.
+    while worker.run_once():
+        pass
+    sender_calls = len(api.sent())
+
+    async def check():
+        app = bot.build(base_url=api.base_url)
+        async with app:
+            # 07:00 Rome: before the daily time, so only the pending "Next" goes out.
+            now = datetime(2026, 11, 2, 7, 0, tzinfo=ZoneInfo("Europe/Rome"))
+            await delivery.deliver_due(sessions, bot_settings, TelegramSender(app.bot, OWNER), now=now)
+
+    asyncio.run(check())
+    message = api.sent()[sender_calls]
+    assert message["parse_mode"] == "HTML"
+    assert message["text"].startswith("The Test Book\nChapter 1 · 1/3")
+    buttons = json.loads(message["reply_markup"])["inline_keyboard"]
+    assert buttons[0][0] == {"text": "Next ▶", "callback_data": f"next:{sub.id}"}
+
+
+def test_reading_pause_and_stop(sessions, bot_settings, worker, api):
+    bot = ChapterwiseBot(bot_settings, sessions)
+    with sessions() as s:
+        epub = s.scalar(select(DocumentRow.id).where(DocumentRow.format == "epub"))
+        sub_id = delivery.subscribe(s, epub).id
+
+    run_updates(bot, api, command("/reading"))
+    assert "Next: section 1 of 3" in api.sent()[0]["text"]
+
+    run_updates(bot, api, tap(f"pause:{sub_id}"))
+    assert api.calls_of("answerCallbackQuery")[-1]["text"] == "Paused"
+    run_updates(bot, api, tap(f"stop:{sub_id}", update_id=2))
+    with sessions() as s:
+        assert s.get(Subscription, sub_id).status == "stopped"
+
+    run_updates(bot, api, tap(f"next:{sub_id}", update_id=3))
+    alert = api.calls_of("answerCallbackQuery")[-1]
+    assert "finished or stopped" in alert["text"] and alert["show_alert"] == "true"
+
+
+def test_rejected_html_falls_back_to_plain_text(sessions, bot_settings, api):
+    api.reject_html = True
+
+    async def go():
+        app = ChapterwiseBot(bot_settings, sessions).build(base_url=api.base_url)
+        async with app:
+            return await TelegramSender(app.bot, OWNER).send(
+                "Book & more", [[("Pause", "cb:pause:1")]]
+            )
+
+    message_id = asyncio.run(go())
+    assert message_id is not None
+    plain = api.sent()[-1]
+    assert "parse_mode" not in plain and plain["text"] == "Book & more"
diff --git a/tests/test_delivery.py b/tests/test_delivery.py
new file mode 100644
index 0000000..6968cb6
--- /dev/null
+++ b/tests/test_delivery.py
@@ -0,0 +1,157 @@
+import asyncio
+from dataclasses import replace
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+import pytest
+from sqlalchemy import select
+
+from chapterwise import delivery
+from chapterwise.db import Delivery, DocumentRow, Job, Subscription
+from chapterwise.worker import Worker
+from tests.conftest import FakeLLM
+
+ROME = ZoneInfo("Europe/Rome")
+
+
+def at(day: int, hour: int, minute: int = 0) -> datetime:
+    return datetime(2026, 11, day, hour, minute, tzinfo=ROME)
+
+
+class FakeSender:
+    def __init__(self):
+        self.messages: list[tuple[str, list]] = []
+        self.notes: list[str] = []
+
+    async def send(self, html, buttons):
+        self.messages.append((html, buttons))
+        return 1000 + len(self.messages)
+
+    async def notify(self, text):
+        self.notes.append(text)
+
+
+@pytest.fixture
+def tg_settings(settings):
+    return replace(
+        settings, telegram_token="123:abc", telegram_chat_id=42, public_url="https://read.example.com"
+    )
+
+
+@pytest.fixture
+def worker(sessions, tg_settings):
+    w = Worker(tg_settings, sessions, llm=FakeLLM("fake-model"))
+    w.schedule_scan()
+    while w.run_once():
+        pass
+    return w
+
+
+@pytest.fixture
+def epub(sessions, worker) -> int:
+    with sessions() as s:
+        return s.scalar(select(DocumentRow.id).where(DocumentRow.format == "epub"))
+
+
+def drain(worker):
+    while worker.run_once():
+        pass
+
+
+def deliver(sessions, settings, now) -> FakeSender:
+    sender = FakeSender()
+    asyncio.run(delivery.deliver_due(sessions, settings, sender, now=now))
+    return sender
+
+
+def test_subscribe_prefetches_and_rejects_bad_input(sessions, epub):
+    with sessions() as s:
+        sub = delivery.subscribe(s, epub)
+        assert (sub.status, sub.position) == ("active", 0)
+        queued = s.scalars(select(Job.payload).where(Job.kind == "summarize")).all()
+        assert sorted(p["position"] for p in queued) == [0, 1, 2]
+        assert delivery.subscribe(s, epub).id == sub.id  # one open subscription per book
+
+        with pytest.raises(delivery.SubscriptionError):
+            delivery.subscribe(s, epub, start=99)
+        with pytest.raises(delivery.SubscriptionError):
+            delivery.subscribe(s, 12345)
+
+
+def test_daily_delivery_at_0730_once_per_day(sessions, tg_settings, worker, epub):
+    with sessions() as s:
+        delivery.subscribe(s, epub)
+    drain(worker)  # summaries for sections 0-2
+
+    assert deliver(sessions, tg_settings, at(2, 7, 29)).messages == []  # too early
+
+    sender = deliver(sessions, tg_settings, at(2, 7, 30))
+    assert len(sender.messages) == 1
+    html, buttons = sender.messages[0]
+    assert html.startswith("The Test Book\nChapter 1 · 1/3")
+    assert "summary" in html
+    assert buttons[0] == [("Next ▶", "cb:next:1"), ("Pause", "cb:pause:1")]
+    assert buttons[1] == [("Open on the web", f"url:https://read.example.com/b/{epub}/s/0")]
+
+    assert deliver(sessions, tg_settings, at(2, 12)).messages == []  # already delivered today
+    assert len(deliver(sessions, tg_settings, at(3, 7, 31)).messages) == 1  # next morning
+
+    with sessions() as s:
+        sub = s.scalar(select(Subscription))
+        assert sub.position == 2 and str(sub.last_delivered_on) == "2026-11-03"
+        assert [d.message_id for d in s.scalars(select(Delivery).order_by(Delivery.id))] == [1001, 1001]
+
+
+def test_missed_morning_is_caught_up(sessions, tg_settings, worker, epub):
+    with sessions() as s:
+        delivery.subscribe(s, epub)
+    drain(worker)
+    # The bot was down at 07:30; the first check afterwards still delivers today's section.
+    assert len(deliver(sessions, tg_settings, at(2, 21, 15)).messages) == 1
+
+
+def test_waits_for_summary_then_delivers(sessions, tg_settings, worker, epub):
+    with sessions() as s:
+        delivery.subscribe(s, epub)
+    assert deliver(sessions, tg_settings, at(2, 8)).messages == []  # summary not ready yet
+    drain(worker)
+    assert len(deliver(sessions, tg_settings, at(2, 8, 1)).messages) == 1
+
+
+def test_next_sends_extra_section_without_skipping_tomorrow(sessions, tg_settings, worker, epub):
+    with sessions() as s:
+        sub_id = delivery.subscribe(s, epub).id
+    drain(worker)
+    deliver(sessions, tg_settings, at(2, 7, 30))  # section 0
+
+    with sessions() as s:
+        delivery.request_next(s, sub_id)
+    drain(worker)
+    manual = deliver(sessions, tg_settings, at(2, 9))
+    assert "Chapter 2 · 2/3" in manual.messages[0][0]
+
+    tomorrow = deliver(sessions, tg_settings, at(3, 7, 30))
+    assert "Chapter 3 · 3/3" in tomorrow.messages[0][0]
+    assert tomorrow.messages[0][1][0] == [("Pause", "cb:pause:1")]  # no Next on the last section
+    assert tomorrow.notes == ["🎉 You finished The Test Book! Pick the next one with /books."]
+    with sessions() as s:
+        assert s.get(Subscription, sub_id).status == "finished"
+    assert deliver(sessions, tg_settings, at(4, 7, 30)).messages == []
+
+
+def test_paused_and_stopped_get_nothing(sessions, tg_settings, worker, epub):
+    with sessions() as s:
+        sub_id = delivery.subscribe(s, epub).id
+        delivery.set_status(s, sub_id, "paused")
+    drain(worker)
+    assert deliver(sessions, tg_settings, at(2, 8)).messages == []
+
+    with sessions() as s:
+        delivery.set_status(s, sub_id, "active")
+    assert len(deliver(sessions, tg_settings, at(2, 8)).messages) == 1
+
+    with sessions() as s:
+        delivery.set_status(s, sub_id, "stopped")
+        with pytest.raises(delivery.SubscriptionError):
+            delivery.request_next(s, sub_id)
+        assert delivery.subscribe(s, epub).id != sub_id  # a stopped book can be started again
diff --git a/tests/test_jobs.py b/tests/test_jobs.py
index ea2476e..ec6d589 100644
--- a/tests/test_jobs.py
+++ b/tests/test_jobs.py
@@ -87,3 +87,12 @@ def test_requeue_stale(sessions):
         jobs.claim(s)
         assert jobs.requeue_stale(s) == 1
         assert jobs.claim(s) is not None
+
+
+def test_enqueue_keeps_working_after_many_executions(sessions):
+    """Regression: a parameterized ON CONFLICT predicate broke once Postgres switched to a generic plan."""
+    with sessions() as s:
+        for n in range(30):
+            assert jobs.enqueue(s, "summarize", {"n": n}, dedupe_key=f"k{n}")
+            assert not jobs.enqueue(s, "summarize", {"n": n}, dedupe_key=f"k{n}")
+        s.commit()
diff --git a/tests/test_telegram_format.py b/tests/test_telegram_format.py
new file mode 100644
index 0000000..32a95a6
--- /dev/null
+++ b/tests/test_telegram_format.py
@@ -0,0 +1,48 @@
+from chapterwise.telegram_format import build_message, markdown_to_blocks, to_plain_text
+
+
+def test_converts_markdown_to_telegram_html():
+    md = """Overview with **bold**, *italic*, `code` and a [link](https://example.com).
+
+## Key ideas
+
+- First **idea**
+- Second idea
+  - nested point
+
+1. one
+2. two
+
+**Takeaway:** keep going."""
+    blocks = markdown_to_blocks(md)
+    assert blocks[0] == (
+        'Overview with bold, italic, code and a link.'
+    )
+    assert blocks[1] == "Key ideas"
+    assert blocks[2] == "• First idea\n• Second idea\n  • nested point"
+    assert blocks[3] == "1. one\n2. two"
+    assert blocks[4] == "Takeaway: keep going."
+
+
+def test_escapes_html_and_drops_unsafe_links():
+    blocks = markdown_to_blocks("Use  & a [bad](javascript:alert(1)) link, 3 < 4")
+    html = " ".join(blocks)
+    assert "