Fix messagesScanned consistency: use totalScanned from scan results, remove double-counting

Co-authored-by: xCyanGrizzly <53275238+xCyanGrizzly@users.noreply.github.com>

Dockerfile

@@ -30,22 +30,23 @@ RUN addgroup --system --gid 1001 nodejs && \
     adduser --system --uid 1001 nextjs
 
 # Copy public assets
-COPY --from=builder /app/public ./public
-
-# Copy prisma schema + migrations for runtime migrate deploy
-COPY --from=builder /app/prisma ./prisma
-COPY --from=builder /app/prisma.config.ts ./prisma.config.ts
+COPY --from=builder --chown=nextjs:nodejs /app/public ./public
 
 # Copy standalone build output
 COPY --from=builder --chown=nextjs:nodejs /app/.next/standalone ./
 COPY --from=builder --chown=nextjs:nodejs /app/.next/static ./.next/static
 
-# Copy node_modules for prisma CLI (needed for migrate deploy at startup)
-COPY --from=builder /app/node_modules/.prisma ./node_modules/.prisma
-COPY --from=builder /app/node_modules/@prisma ./node_modules/@prisma
-COPY --from=builder /app/node_modules/prisma ./node_modules/prisma
-COPY --from=builder /app/node_modules/.bin/prisma ./node_modules/.bin/prisma
-COPY --from=builder /app/node_modules/dotenv ./node_modules/dotenv
+# Copy prisma schema + migrations for runtime migrate deploy
+COPY --from=builder --chown=nextjs:nodejs /app/prisma ./prisma
+COPY --from=builder --chown=nextjs:nodejs /app/prisma.config.ts ./prisma.config.ts
+
+# Copy node_modules for prisma CLI (needed for migrate deploy at startup).
+# Copying the full directory ensures all transitive dependencies are present.
+COPY --from=builder --chown=nextjs:nodejs /app/node_modules ./node_modules
+# Recreate the .bin/prisma symlink so Node resolves __dirname to prisma/build/,
+# where the WASM files live (COPY dereferences symlinks, breaking WASM resolution)
+RUN mkdir -p ./node_modules/.bin && \
+    ln -sf ../prisma/build/index.js ./node_modules/.bin/prisma
 
 # Copy entrypoint script
 COPY --chown=nextjs:nodejs docker-entrypoint.sh ./

README.md

@@ -125,18 +125,15 @@ docker compose up -d
 
 The app will be available at [http://localhost:3000](http://localhost:3000).
 
-### Adding Telegram Services
+### Adding the Telegram Bot
 
-The worker and bot run as optional profiles so `docker compose up` works with just the app + database:
+The worker starts by default with `docker compose up`. The bot runs as an optional profile:
 
 ```bash
-# App + DB + Telegram worker (needs TELEGRAM_API_ID + TELEGRAM_API_HASH in .env)
-docker compose --profile telegram up -d
-
 # App + DB + Worker + Bot (also needs BOT_TOKEN in .env)
 docker compose --profile full up -d
 
-# Or just the bot (alongside app + db)
+# Or just the bot (alongside app + db + worker)
 docker compose --profile bot up -d
 ```
 

docker-compose.dev.yml

@@ -16,7 +16,6 @@ services:
       retries: 5
 
   worker:
-    profiles: ["telegram", "full"]
     build:
       context: .
       dockerfile: worker/Dockerfile

docker-compose.yml

@@ -3,15 +3,20 @@ services:
     build:
       context: .
       dockerfile: Dockerfile
+    pull_policy: never
     ports:
       - "${APP_PORT:-3000}:3000"
     environment:
       - DATABASE_URL=postgresql://${POSTGRES_USER:-dragons}:${POSTGRES_PASSWORD:-stash}@db:5432/${POSTGRES_DB:-dragonsstash}
       - AUTH_SECRET=${AUTH_SECRET:?Set AUTH_SECRET in .env}
       - AUTH_TRUST_HOST=true
+      - AUTH_GITHUB_ID=${AUTH_GITHUB_ID:-}
+      - AUTH_GITHUB_SECRET=${AUTH_GITHUB_SECRET:-}
       - NEXT_PUBLIC_APP_URL=${NEXT_PUBLIC_APP_URL:-http://localhost:3000}
       - TELEGRAM_API_KEY=${TELEGRAM_API_KEY:-}
       - BOT_TOKEN=${BOT_TOKEN:-}
+      - BOT_USERNAME=${BOT_USERNAME:-}
+      - LOG_LEVEL=${LOG_LEVEL:-info}
     depends_on:
       db:
         condition: service_healthy
@@ -20,7 +25,7 @@ services:
       interval: 30s
       timeout: 5s
       retries: 3
-      start_period: 30s
+      start_period: 60s
     restart: unless-stopped
     deploy:
       resources:
@@ -30,14 +35,14 @@ services:
       - frontend
 
   worker:
-    profiles: ["telegram", "full"]
     build:
       context: .
       dockerfile: worker/Dockerfile
+    pull_policy: never
     environment:
       - DATABASE_URL=postgresql://${POSTGRES_USER:-dragons}:${POSTGRES_PASSWORD:-stash}@db:5432/${POSTGRES_DB:-dragonsstash}
-      - TELEGRAM_API_ID=${TELEGRAM_API_ID:?Set TELEGRAM_API_ID in .env}
-      - TELEGRAM_API_HASH=${TELEGRAM_API_HASH:?Set TELEGRAM_API_HASH in .env}
+      - TELEGRAM_API_ID=${TELEGRAM_API_ID:-}
+      - TELEGRAM_API_HASH=${TELEGRAM_API_HASH:-}
       - WORKER_INTERVAL_MINUTES=${WORKER_INTERVAL_MINUTES:-60}
       - WORKER_TEMP_DIR=/tmp/zips
       - TDLIB_STATE_DIR=/data/tdlib
@@ -65,11 +70,12 @@ services:
     build:
       context: .
       dockerfile: bot/Dockerfile
+    pull_policy: never
     environment:
       - DATABASE_URL=postgresql://${POSTGRES_USER:-dragons}:${POSTGRES_PASSWORD:-stash}@db:5432/${POSTGRES_DB:-dragonsstash}
-      - BOT_TOKEN=${BOT_TOKEN:?Set BOT_TOKEN in .env}
-      - TELEGRAM_API_ID=${TELEGRAM_API_ID:?Set TELEGRAM_API_ID in .env}
-      - TELEGRAM_API_HASH=${TELEGRAM_API_HASH:?Set TELEGRAM_API_HASH in .env}
+      - BOT_TOKEN=${BOT_TOKEN:-}
+      - TELEGRAM_API_ID=${TELEGRAM_API_ID:-}
+      - TELEGRAM_API_HASH=${TELEGRAM_API_HASH:-}
       - LOG_LEVEL=${LOG_LEVEL:-info}
     volumes:
       - tdlib_bot_state:/data/tdlib

docker-entrypoint.sh

@@ -10,11 +10,14 @@ if [ "$AUTH_SECRET" = "change-me-to-a-random-secret-in-production" ] || [ -z "$A
 fi
 
 echo "Running database migrations..."
-npx prisma migrate deploy
+if ! ./node_modules/.bin/prisma migrate deploy; then
+  echo "ERROR: Database migration failed. Check DATABASE_URL and database connectivity."
+  exit 1
+fi
 
 if [ "$SEED_DATABASE" = "true" ]; then
   echo "Seeding database..."
-  npx prisma db seed || echo "Seeding skipped or already done."
+  ./node_modules/.bin/prisma db seed || echo "Seeding skipped or already done."
 fi
 
 echo "Starting application..."

package-lock.json

@@ -19,6 +19,7 @@
         "clsx": "^2.1.1",
         "cmdk": "^1.1.1",
         "dotenv": "^17.3.1",
+        "geist": "^1.7.0",
         "lucide-react": "^0.574.0",
         "next": "16.1.6",
         "next-auth": "^5.0.0-beta.30",
@@ -8100,6 +8101,15 @@
       "dev": true,
       "license": "BSD-3-Clause"
     },
+    "node_modules/geist": {
+      "version": "1.7.0",
+      "resolved": "https://registry.npmjs.org/geist/-/geist-1.7.0.tgz",
+      "integrity": "sha512-ZaoiZwkSf0DwwB1ncdLKp+ggAldqxl5L1+SXaNIBGkPAqcu+xjVJLxlf3/S8vLt9UHx1xu5fz3lbzKCj5iOVdQ==",
+      "license": "SIL OPEN FONT LICENSE",
+      "peerDependencies": {
+        "next": ">=13.2.0"
+      }
+    },
     "node_modules/generate-function": {
       "version": "2.3.1",
       "resolved": "https://registry.npmjs.org/generate-function/-/generate-function-2.3.1.tgz",

package.json

@@ -28,6 +28,7 @@
     "clsx": "^2.1.1",
     "cmdk": "^1.1.1",
     "dotenv": "^17.3.1",
+    "geist": "^1.7.0",
     "lucide-react": "^0.574.0",
     "next": "16.1.6",
     "next-auth": "^5.0.0-beta.30",

prisma/migrations/20260304200000_make_all_users_admin/migration.sql

@@ -0,0 +1,5 @@
+-- Promote all existing users to ADMIN (self-hosted: every user is an admin)
+UPDATE "User" SET "role" = 'ADMIN' WHERE "role" = 'USER';
+
+-- Change the default role for new users to ADMIN
+ALTER TABLE "User" ALTER COLUMN "role" SET DEFAULT 'ADMIN';

prisma/migrations/20260305200000_default_channels_disabled/migration.sql

@@ -0,0 +1,3 @@
+-- Change the default for new channels to disabled (isActive = false).
+-- Existing channels are not affected — admins can manually enable/disable them.
+ALTER TABLE "telegram_channels" ALTER COLUMN "isActive" SET DEFAULT false;

prisma/schema.prisma

@@ -22,7 +22,7 @@ model User {
   emailVerified  DateTime?
   image          String?
   hashedPassword String?
-  role           Role      @default(USER)
+  role           Role      @default(ADMIN)
   createdAt      DateTime  @default(now())
   updatedAt      DateTime  @updatedAt
 
@@ -417,7 +417,7 @@ model TelegramChannel {
   title      String
   type       ChannelType
   isForum    Boolean     @default(false)
-  isActive   Boolean     @default(true)
+  isActive   Boolean     @default(false)
   createdAt  DateTime    @default(now())
   updatedAt  DateTime    @updatedAt
 

src/app/(app)/telegram/_components/worker-status-panel.tsx

@@ -233,6 +233,11 @@ function RunningStatus({
             </span>
           </span>
         )}
+        {run.messagesScanned > 0 && (
+          <span>
+            <span className="text-foreground tabular-nums">{run.messagesScanned}</span> messages
+          </span>
+        )}
         {run.zipsIngested > 0 && (
           <span>
             <span className="text-foreground tabular-nums">{run.zipsIngested}</span> ingested

src/app/(app)/telegram/actions.ts

@@ -173,6 +173,7 @@ export async function createChannel(
         telegramId: BigInt(parsed.data.telegramId),
         title: parsed.data.title,
         type: parsed.data.type,
+        isActive: false,
       },
     });
     revalidatePath(REVALIDATE_PATH);
@@ -371,19 +372,8 @@ export async function triggerIngestion(
       return { success: false, error: "No eligible accounts found" };
     }
 
-    // Create ingestion runs — the worker picks these up
-    for (const account of accounts) {
-      const existing = await prisma.ingestionRun.findFirst({
-        where: { accountId: account.id, status: "RUNNING" },
-      });
-      if (!existing) {
-        await prisma.ingestionRun.create({
-          data: { accountId: account.id, status: "RUNNING" },
-        });
-      }
-    }
-
-    // pg_notify for immediate worker pickup
+    // Signal the worker to run an immediate ingestion cycle via pg_notify.
+    // The worker will create its own IngestionRun records with proper activity tracking.
     try {
       await prisma.$queryRawUnsafe(
         `SELECT pg_notify('ingestion_trigger', $1)`,
@@ -417,7 +407,7 @@ export async function saveChannelSelections(
   try {
     let linked = 0;
     for (const ch of channels) {
-      // Upsert the channel record
+      // Upsert the channel record (new channels default to disabled)
       const channel = await prisma.telegramChannel.upsert({
         where: { telegramId: BigInt(ch.telegramId) },
         create: {
@@ -425,6 +415,7 @@ export async function saveChannelSelections(
           title: ch.title,
           type: "SOURCE",
           isForum: ch.isForum,
+          isActive: false,
         },
         update: {
           title: ch.title,
@@ -467,10 +458,10 @@ export async function setGlobalDestination(
   if (!channel) return { success: false, error: "Channel not found" };
 
   try {
-    // Set the channel type to DESTINATION
+    // Set the channel type to DESTINATION and ensure it's active
     await prisma.telegramChannel.update({
       where: { id: channelId },
-      data: { type: "DESTINATION" },
+      data: { type: "DESTINATION", isActive: true },
     });
 
     // Save as global destination
@@ -521,17 +512,19 @@ export async function createDestinationChannel(
   if (!admin.success) return admin;
 
   try {
-    // Create the channel as DESTINATION
+    // Create the channel as DESTINATION (active by default — needed for uploads)
     const channel = await prisma.telegramChannel.upsert({
       where: { telegramId: BigInt(telegramId) },
       create: {
         telegramId: BigInt(telegramId),
         title,
         type: "DESTINATION",
+        isActive: true,
       },
       update: {
         title,
         type: "DESTINATION",
+        isActive: true,
       },
     });
 

src/app/(auth)/register/actions.ts

@@ -21,12 +21,13 @@ export async function registerUser(input: unknown): Promise<ActionResult<{ id: s
 
   const hashedPassword = await bcrypt.hash(parsed.data.password, 10);
 
+  // Self-hosted: all users are admins
   const user = await prisma.user.create({
     data: {
       name: parsed.data.name,
       email: parsed.data.email,
       hashedPassword,
-      role: "USER",
+      role: "ADMIN",
       settings: {
         create: {
           lowStockThreshold: 10,

src/app/layout.tsx

@@ -1,5 +1,6 @@
 import type { Metadata } from "next";
-import { Geist, Geist_Mono } from "next/font/google";
+import { GeistSans } from "geist/font/sans";
+import { GeistMono } from "geist/font/mono";
 import { SessionProvider } from "@/components/providers/session-provider";
 import { ThemeProvider } from "@/components/providers/theme-provider";
 import { Toaster } from "@/components/ui/sonner";
@@ -7,16 +8,6 @@ import { TooltipProvider } from "@/components/ui/tooltip";
 import { APP_NAME } from "@/lib/constants";
 import "./globals.css";
 
-const geistSans = Geist({
-  variable: "--font-geist-sans",
-  subsets: ["latin"],
-});
-
-const geistMono = Geist_Mono({
-  variable: "--font-geist-mono",
-  subsets: ["latin"],
-});
-
 export const metadata: Metadata = {
   title: {
     default: APP_NAME,
@@ -29,7 +20,7 @@ export const metadata: Metadata = {
 export default function RootLayout({ children }: { children: React.ReactNode }) {
   return (
     <html lang="en" className="dark" suppressHydrationWarning>
-      <body className={`${geistSans.variable} ${geistMono.variable} font-sans antialiased`}>
+      <body className={`${GeistSans.variable} ${GeistMono.variable} font-sans antialiased`}>
         <SessionProvider>
           <ThemeProvider>
             <TooltipProvider delayDuration={0}>

src/components/layout/mobile-sidebar.tsx

@@ -2,6 +2,7 @@
 
 import Link from "next/link";
 import { usePathname } from "next/navigation";
+import { useSession } from "next-auth/react";
 import {
   LayoutDashboard,
   Cylinder,
@@ -17,27 +18,17 @@ import {
   Flame,
 } from "lucide-react";
 import { cn } from "@/lib/utils";
-import { APP_NAME } from "@/lib/constants";
+import { APP_NAME, NAV_ITEMS } from "@/lib/constants";
 import { SheetHeader, SheetTitle } from "@/components/ui/sheet";
 
 const icons = { LayoutDashboard, Cylinder, Droplets, Paintbrush, Gem, FileBox, Send, ClipboardList, Building2, MapPin, Settings };
 
-const navItems = [
-  { label: "Dashboard", href: "/dashboard", icon: "LayoutDashboard" as const },
-  { label: "Filaments", href: "/filaments", icon: "Cylinder" as const },
-  { label: "Resins", href: "/resins", icon: "Droplets" as const },
-  { label: "Paints", href: "/paints", icon: "Paintbrush" as const },
-  { label: "Supplies", href: "/supplies", icon: "Gem" as const },
-  { label: "STL Files", href: "/stls", icon: "FileBox" as const },
-  { label: "Telegram", href: "/telegram", icon: "Send" as const },
-  { label: "Usage", href: "/usage", icon: "ClipboardList" as const },
-  { label: "Vendors", href: "/vendors", icon: "Building2" as const },
-  { label: "Locations", href: "/locations", icon: "MapPin" as const },
-  { label: "Settings", href: "/settings", icon: "Settings" as const },
-];
-
 export function MobileSidebar() {
   const pathname = usePathname();
+  const { data: session } = useSession();
+  const isAdmin = session?.user?.role === "ADMIN";
+
+  const visibleItems = NAV_ITEMS.filter((item) => !item.adminOnly || isAdmin);
 
   return (
     <div className="flex h-full flex-col">
@@ -48,7 +39,7 @@ export function MobileSidebar() {
         </SheetTitle>
       </SheetHeader>
       <nav className="flex-1 space-y-1 p-2">
-        {navItems.map((item) => {
+        {visibleItems.map((item) => {
           const Icon = icons[item.icon];
           const isActive = pathname.startsWith(item.href);
 

src/components/layout/sidebar.tsx

@@ -3,6 +3,7 @@
 import { useState } from "react";
 import Link from "next/link";
 import { usePathname } from "next/navigation";
+import { useSession } from "next-auth/react";
 import {
   LayoutDashboard,
   Cylinder,
@@ -20,7 +21,7 @@ import {
   PanelLeft,
 } from "lucide-react";
 import { cn } from "@/lib/utils";
-import { APP_NAME } from "@/lib/constants";
+import { APP_NAME, NAV_ITEMS } from "@/lib/constants";
 import { Button } from "@/components/ui/button";
 import { Tooltip, TooltipContent, TooltipTrigger } from "@/components/ui/tooltip";
 
@@ -38,23 +39,13 @@ const icons = {
   Settings,
 } as const;
 
-const navItems = [
-  { label: "Dashboard", href: "/dashboard", icon: "LayoutDashboard" as const },
-  { label: "Filaments", href: "/filaments", icon: "Cylinder" as const },
-  { label: "Resins", href: "/resins", icon: "Droplets" as const },
-  { label: "Paints", href: "/paints", icon: "Paintbrush" as const },
-  { label: "Supplies", href: "/supplies", icon: "Gem" as const },
-  { label: "STL Files", href: "/stls", icon: "FileBox" as const },
-  { label: "Telegram", href: "/telegram", icon: "Send" as const },
-  { label: "Usage", href: "/usage", icon: "ClipboardList" as const },
-  { label: "Vendors", href: "/vendors", icon: "Building2" as const },
-  { label: "Locations", href: "/locations", icon: "MapPin" as const },
-  { label: "Settings", href: "/settings", icon: "Settings" as const },
-];
-
 export function Sidebar() {
   const pathname = usePathname();
   const [collapsed, setCollapsed] = useState(false);
+  const { data: session } = useSession();
+  const isAdmin = session?.user?.role === "ADMIN";
+
+  const visibleItems = NAV_ITEMS.filter((item) => !item.adminOnly || isAdmin);
 
   return (
     <aside
@@ -73,7 +64,7 @@ export function Sidebar() {
 
       {/* Navigation */}
       <nav className="flex-1 space-y-1 p-2">
-        {navItems.map((item) => {
+        {visibleItems.map((item) => {
           const Icon = icons[item.icon];
           const isActive = pathname.startsWith(item.href);
 

src/lib/auth.ts

@@ -18,7 +18,12 @@ export const { auth, handlers, signIn, signOut } = NextAuth({
     async jwt({ token, user }) {
       if (user) {
         token.id = user.id!;
-        token.role = user.role ?? "USER";
+        // Fetch the role from the database to ensure token reflects current role
+        const dbUser = await prisma.user.findUnique({
+          where: { id: user.id! },
+          select: { role: true },
+        });
+        token.role = dbUser?.role ?? user.role ?? "ADMIN";
       }
       return token;
     },
@@ -33,6 +38,12 @@ export const { auth, handlers, signIn, signOut } = NextAuth({
   events: {
     async createUser({ user }) {
       if (user.id) {
+        // Self-hosted: all users are admins
+        await prisma.user.update({
+          where: { id: user.id },
+          data: { role: "ADMIN" },
+        });
+
         await prisma.userSettings.upsert({
           where: { userId: user.id },
           update: {},

src/lib/constants.ts

@@ -1,17 +1,17 @@
 export const APP_NAME = "Dragon's Stash";
 
 export const NAV_ITEMS = [
-  { label: "Dashboard", href: "/dashboard", icon: "LayoutDashboard" },
-  { label: "Filaments", href: "/filaments", icon: "Cylinder" },
-  { label: "Resins", href: "/resins", icon: "Droplets" },
-  { label: "Paints", href: "/paints", icon: "Paintbrush" },
-  { label: "Supplies", href: "/supplies", icon: "Gem" },
-  { label: "STL Files", href: "/stls", icon: "FileBox" },
-  { label: "Telegram", href: "/telegram", icon: "Send" },
-  { label: "Usage", href: "/usage", icon: "ClipboardList" },
-  { label: "Vendors", href: "/vendors", icon: "Building2" },
-  { label: "Locations", href: "/locations", icon: "MapPin" },
-  { label: "Settings", href: "/settings", icon: "Settings" },
+  { label: "Dashboard", href: "/dashboard", icon: "LayoutDashboard", adminOnly: false },
+  { label: "Filaments", href: "/filaments", icon: "Cylinder", adminOnly: false },
+  { label: "Resins", href: "/resins", icon: "Droplets", adminOnly: false },
+  { label: "Paints", href: "/paints", icon: "Paintbrush", adminOnly: false },
+  { label: "Supplies", href: "/supplies", icon: "Gem", adminOnly: false },
+  { label: "STL Files", href: "/stls", icon: "FileBox", adminOnly: false },
+  { label: "Telegram", href: "/telegram", icon: "Send", adminOnly: true },
+  { label: "Usage", href: "/usage", icon: "ClipboardList", adminOnly: false },
+  { label: "Vendors", href: "/vendors", icon: "Building2", adminOnly: false },
+  { label: "Locations", href: "/locations", icon: "MapPin", adminOnly: false },
+  { label: "Settings", href: "/settings", icon: "Settings", adminOnly: false },
 ] as const;
 
 export const MATERIALS = [

src/lib/fonts.ts

@@ -1,6 +0,0 @@
-import { Inter } from "next/font/google";
-
-export const inter = Inter({
-  subsets: ["latin"],
-  variable: "--font-inter",
-});

worker/src/db/queries.ts

@@ -302,11 +302,15 @@ export interface UpsertChannelInput {
   title: string;
   type: "SOURCE" | "DESTINATION";
   isForum: boolean;
+  isActive?: boolean;
 }
 
 /**
  * Upsert a channel by telegramId. Returns the channel record.
  * If it already exists, update title and forum status.
+ * New channels default to disabled (isActive: false) so the admin must
+ * explicitly enable them before the worker processes them.
+ * Pass isActive: true for DESTINATION channels that must be active immediately.
  */
 export async function upsertChannel(input: UpsertChannelInput) {
   return db.telegramChannel.upsert({
@@ -316,6 +320,7 @@ export async function upsertChannel(input: UpsertChannelInput) {
       title: input.title,
       type: input.type,
       isForum: input.isForum,
+      isActive: input.isActive ?? false,
     },
     update: {
       title: input.title,

worker/src/fetch-listener.ts

@@ -5,6 +5,7 @@ import { withTdlibMutex } from "./util/mutex.js";
 import { processFetchRequest } from "./worker.js";
 import { generateInviteLink, createSupergroup } from "./tdlib/chats.js";
 import { createTdlibClient, closeTdlibClient } from "./tdlib/client.js";
+import { triggerImmediateCycle } from "./scheduler.js";
 import {
   getGlobalDestinationChannel,
   getGlobalSetting,
@@ -25,12 +26,14 @@ let pgClient: pg.PoolClient | null = null;
  *   - `channel_fetch` — payload = requestId → fetch channels for an account
  *   - `generate_invite` — payload = channelId → generate invite link for destination
  *   - `create_destination` — payload = JSON { requestId, title } → create supergroup via TDLib
+ *   - `ingestion_trigger` — trigger an immediate ingestion cycle
  */
 export async function startFetchListener(): Promise<void> {
   pgClient = await pool.connect();
   await pgClient.query("LISTEN channel_fetch");
   await pgClient.query("LISTEN generate_invite");
   await pgClient.query("LISTEN create_destination");
+  await pgClient.query("LISTEN ingestion_trigger");
 
   pgClient.on("notification", (msg) => {
     if (msg.channel === "channel_fetch" && msg.payload) {
@@ -39,10 +42,12 @@ export async function startFetchListener(): Promise<void> {
       handleGenerateInvite(msg.payload);
     } else if (msg.channel === "create_destination" && msg.payload) {
       handleCreateDestination(msg.payload);
+    } else if (msg.channel === "ingestion_trigger") {
+      handleIngestionTrigger();
     }
   });
 
-  log.info("Fetch listener started (channel_fetch, generate_invite, create_destination)");
+  log.info("Fetch listener started (channel_fetch, generate_invite, create_destination, ingestion_trigger)");
 }
 
 export function stopFetchListener(): void {
@@ -138,12 +143,13 @@ function handleCreateDestination(payload: string): void {
           const result = await createSupergroup(client, parsed.title);
           log.info({ chatId: result.chatId.toString(), title: result.title }, "Supergroup created");
 
-          // Upsert it as a DESTINATION channel in the DB
+          // Upsert it as a DESTINATION channel in the DB (active by default)
           const channel = await upsertChannel({
             telegramId: result.chatId,
             title: result.title,
             type: "DESTINATION",
             isForum: false,
+            isActive: true,
           });
 
           // Set as global destination
@@ -204,3 +210,16 @@ function handleCreateDestination(payload: string): void {
     }
   });
 }
+
+// ── Ingestion trigger handler ──
+
+function handleIngestionTrigger(): void {
+  fetchQueue = fetchQueue.then(async () => {
+    try {
+      log.info("Ingestion trigger received from UI");
+      await triggerImmediateCycle();
+    } catch (err) {
+      log.error({ err }, "Failed to trigger immediate ingestion cycle");
+    }
+  });
+}

worker/src/index.ts

@@ -13,6 +13,11 @@ async function main(): Promise<void> {
   log.info("DragonsStash Telegram Worker starting");
   log.info({ config: { ...config, databaseUrl: "***" } }, "Configuration loaded");
 
+  if (!config.telegramApiId || !config.telegramApiHash) {
+    log.fatal("TELEGRAM_API_ID and TELEGRAM_API_HASH are both required");
+    process.exit(1);
+  }
+
   // Ensure temp directory exists
   await mkdir(config.tempDir, { recursive: true });
   await mkdir(config.tdlibStateDir, { recursive: true });

worker/src/scheduler.ts

@@ -105,6 +105,19 @@ export async function startScheduler(): Promise<void> {
   scheduleNext();
 }
 
+/**
+ * Trigger an immediate ingestion cycle (e.g. from the admin UI).
+ * If a cycle is already running, this is a no-op.
+ */
+export async function triggerImmediateCycle(): Promise<void> {
+  if (running) {
+    log.info("Cycle already running, ignoring trigger");
+    return;
+  }
+  log.info("Immediate cycle triggered via UI");
+  await runCycle();
+}
+
 /**
  * Stop the scheduler gracefully.
  */

worker/src/tdlib/download.ts

@@ -66,8 +66,11 @@ interface TdFile {
 export interface ChannelScanResult {
   archives: TelegramMessage[];
   photos: TelegramPhoto[];
+  totalScanned: number;
 }
 
+export type ScanProgressCallback = (messagesScanned: number) => void;
+
 /**
  * Fetch messages from a channel, stopping once we've scanned past the
  * last-processed boundary (with one page of lookback for multipart safety).
@@ -82,13 +85,15 @@ export async function getChannelMessages(
   client: Client,
   chatId: bigint,
   lastProcessedMessageId?: bigint | null,
-  limit = 100
+  limit = 100,
+  onProgress?: ScanProgressCallback
 ): Promise<ChannelScanResult> {
   const archives: TelegramMessage[] = [];
   const photos: TelegramPhoto[] = [];
   const boundary = lastProcessedMessageId ? Number(lastProcessedMessageId) : null;
 
   let currentFromId = 0;
+  let totalScanned = 0;
 
   // eslint-disable-next-line no-constant-condition
   while (true) {
@@ -103,6 +108,8 @@ export async function getChannelMessages(
 
     if (!result.messages || result.messages.length === 0) break;
 
+    totalScanned += result.messages.length;
+
     for (const msg of result.messages) {
       // Check for archive documents
       const doc = msg.content?.document;
@@ -132,6 +139,9 @@ export async function getChannelMessages(
       }
     }
 
+    // Report scanning progress after each page
+    onProgress?.(totalScanned);
+
     currentFromId = result.messages[result.messages.length - 1].id;
 
     // Stop scanning once we've gone past the boundary (this page is the lookback)
@@ -144,7 +154,7 @@ export async function getChannelMessages(
   }
 
   log.info(
-    { chatId: chatId.toString(), archives: archives.length, photos: photos.length },
+    { chatId: chatId.toString(), archives: archives.length, photos: photos.length, totalScanned },
     "Channel scan complete"
   );
 
@@ -152,6 +162,7 @@ export async function getChannelMessages(
   return {
     archives: archives.reverse(),
     photos: photos.reverse(),
+    totalScanned,
   };
 }
 

worker/src/tdlib/topics.ts

@@ -4,7 +4,7 @@ import { childLogger } from "../util/logger.js";
 import { isArchiveAttachment } from "../archive/detect.js";
 import type { TelegramMessage } from "../archive/multipart.js";
 import type { TelegramPhoto } from "../preview/match.js";
-import type { ChannelScanResult } from "./download.js";
+import type { ChannelScanResult, ScanProgressCallback } from "./download.js";
 
 const log = childLogger("topics");
 
@@ -140,13 +140,15 @@ export async function getTopicMessages(
   chatId: bigint,
   topicId: bigint,
   lastProcessedMessageId?: bigint | null,
-  limit = 100
+  limit = 100,
+  onProgress?: ScanProgressCallback
 ): Promise<ChannelScanResult> {
   const archives: TelegramMessage[] = [];
   const photos: TelegramPhoto[] = [];
   const boundary = lastProcessedMessageId ? Number(lastProcessedMessageId) : null;
 
   let currentFromId = 0;
+  let totalScanned = 0;
 
   // eslint-disable-next-line no-constant-condition
   while (true) {
@@ -190,6 +192,8 @@ export async function getTopicMessages(
 
     if (!result.messages || result.messages.length === 0) break;
 
+    totalScanned += result.messages.length;
+
     for (const msg of result.messages) {
       // Check for archive documents
       const doc = msg.content?.document;
@@ -219,6 +223,9 @@ export async function getTopicMessages(
       }
     }
 
+    // Report scanning progress after each page
+    onProgress?.(totalScanned);
+
     currentFromId = result.messages[result.messages.length - 1].id;
 
     // Stop scanning once we've gone past the boundary (this page is the lookback)
@@ -230,7 +237,7 @@ export async function getTopicMessages(
   }
 
   log.info(
-    { chatId: chatId.toString(), topicId: topicId.toString(), archives: archives.length, photos: photos.length },
+    { chatId: chatId.toString(), topicId: topicId.toString(), archives: archives.length, photos: photos.length, totalScanned },
     "Topic scan complete"
   );
 
@@ -238,6 +245,7 @@ export async function getTopicMessages(
   return {
     archives: archives.reverse(),
     photos: photos.reverse(),
+    totalScanned,
   };
 }
 

worker/src/worker.ts

@@ -349,8 +349,14 @@ export async function runWorkerForAccount(
         throw new Error("No global destination channel configured — set one in the admin UI");
       }
 
-      for (const mapping of channelMappings) {
+      const totalChannels = channelMappings.length;
+
+      for (let chIdx = 0; chIdx < channelMappings.length; chIdx++) {
+        const mapping = channelMappings[chIdx];
         const channel = mapping.channel;
+        const channelLabel = totalChannels > 1
+          ? `[${chIdx + 1}/${totalChannels}] ${channel.title}`
+          : channel.title;
 
         try {
         // ── Check if channel is a forum ──
@@ -380,15 +386,16 @@ export async function runWorkerForAccount(
         if (forum) {
           // ── Forum channel: scan per-topic ──
           await updateRunActivity(activeRunId, {
-            currentActivity: `Enumerating topics in "${channel.title}"`,
+            currentActivity: `Enumerating topics in "${channelLabel}"`,
             currentStep: "scanning",
-            currentChannel: channel.title,
+            currentChannel: channelLabel,
             currentFile: null,
             currentFileNum: null,
             totalFiles: null,
             downloadedBytes: null,
             totalBytes: null,
             downloadPercent: null,
+            messagesScanned: counters.messagesScanned,
           });
 
           const topics = await getForumTopicList(client, channel.telegramId);
@@ -399,31 +406,50 @@ export async function runWorkerForAccount(
             "Scanning forum channel by topic"
           );
 
-          for (const topic of topics) {
+          for (let tIdx = 0; tIdx < topics.length; tIdx++) {
+            const topic = topics[tIdx];
             try {
               const progress = topicProgressList.find(
                 (tp) => tp.topicId === topic.topicId
               );
 
+              const topicLabel = `${channel.title} › ${topic.name}`;
+              const topicProgress = topics.length > 1
+                ? ` (topic ${tIdx + 1}/${topics.length})`
+                : "";
+
               await updateRunActivity(activeRunId, {
-                currentActivity: `Scanning topic "${topic.name}" in "${channel.title}"`,
+                currentActivity: `Scanning "${topicLabel}"${topicProgress}`,
                 currentStep: "scanning",
-                currentChannel: `${channel.title} › ${topic.name}`,
+                currentChannel: channelLabel,
                 currentFile: null,
                 currentFileNum: null,
                 totalFiles: null,
                 downloadedBytes: null,
                 totalBytes: null,
                 downloadPercent: null,
+                messagesScanned: counters.messagesScanned,
               });
 
               const scanResult = await getTopicMessages(
                 client,
                 channel.telegramId,
                 topic.topicId,
-                progress?.lastProcessedMessageId
+                progress?.lastProcessedMessageId,
+                100,
+                (scanned) => {
+                  throttled.update({
+                    currentActivity: `Scanning "${topicLabel}"${topicProgress} — ${scanned} messages scanned`,
+                    currentStep: "scanning",
+                    currentChannel: channelLabel,
+                    messagesScanned: counters.messagesScanned + scanned,
+                  });
+                }
               );
 
+              // Add scanned messages to global counter
+              counters.messagesScanned += scanResult.totalScanned;
+
               if (scanResult.archives.length === 0) {
                 accountLog.debug(
                   { channelId: channel.id, topic: topic.name },
@@ -463,15 +489,16 @@ export async function runWorkerForAccount(
         } else {
           // ── Non-forum channel: flat scan (existing behavior) ──
           await updateRunActivity(activeRunId, {
-            currentActivity: `Scanning "${channel.title}" for new archives`,
+            currentActivity: `Scanning "${channelLabel}" for new archives`,
             currentStep: "scanning",
-            currentChannel: channel.title,
+            currentChannel: channelLabel,
             currentFile: null,
             currentFileNum: null,
             totalFiles: null,
             downloadedBytes: null,
             totalBytes: null,
             downloadPercent: null,
+            messagesScanned: counters.messagesScanned,
           });
 
           accountLog.info(
@@ -482,9 +509,21 @@ export async function runWorkerForAccount(
           const scanResult = await getChannelMessages(
             client,
             channel.telegramId,
-            mapping.lastProcessedMessageId
+            mapping.lastProcessedMessageId,
+            100,
+            (scanned) => {
+              throttled.update({
+                currentActivity: `Scanning "${channelLabel}" — ${scanned} messages scanned`,
+                currentStep: "scanning",
+                currentChannel: channelLabel,
+                messagesScanned: counters.messagesScanned + scanned,
+              });
+            }
           );
 
+          // Add scanned messages to global counter
+          counters.messagesScanned += scanResult.totalScanned;
+
           if (scanResult.archives.length === 0) {
             accountLog.debug({ channelId: channel.id }, "No new archives");
             continue;
@@ -593,6 +632,7 @@ async function processArchiveSets(
     currentChannel: channelTitle,
     totalFiles: archiveSets.length,
     zipsFound: counters.zipsFound,
+    messagesScanned: counters.messagesScanned,
   });
 
   // Track the highest message ID that was successfully processed
@@ -646,7 +686,6 @@ async function processOneArchiveSet(
     throttled, counters, topicCreator, sourceTopicId, accountLog,
   } = ctx;
 
-  counters.messagesScanned += archiveSet.parts.length;
   const archiveName = archiveSet.parts[0].fileName;
 
   // ── Early skip: check if this archive set was already ingested ──