10 Commits

Author SHA1 Message Date
6178ff3b08 docs: add Task 3 (creator filter combobox) to plan
All checks were successful
continuous-integration/drone/push Build is passing
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-04 09:42:37 +02:00
57842c6d95 Merge feat/send-all-from-creator: send-all-from-creator button + searchable creator filter 2026-07-04 09:42:08 +02:00
b3c49c3794 feat(stls): add searchable creator filter combobox to toolbar 2026-07-04 09:36:51 +02:00
20e60bc6af feat(stls): add 'send all from creator' toolbar button 2026-07-04 09:26:36 +02:00
50e89719bb feat(stls): add sendAllFromCreatorAction server action 2026-07-04 09:20:36 +02:00
1cae855c26 docs: implementation plan for send-all-from-creator button
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-04 09:15:58 +02:00
b0baf72f0a docs: spec for send-all-from-creator button on STL page
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-04 09:14:33 +02:00
0cf5fcd3a7 fix(stls): stop loading preview blobs in list queries (OOM fix)
All checks were successful
continuous-integration/drone/push Build is passing
The STLs list queries selected the full previewData JPEG bytes (avg
~700KB, up to 2MB, and every group member's preview) only to compute a
hasPreview boolean. Under concurrent page loads this piled blobs into
the Node heap, exceeded the 512MB container limit, crashed the process,
and surfaced as repeated connection errors while browsing.

Replace the byte-loading select with a fetchPreviewFlags() helper that
checks `previewData IS NOT NULL` in SQL (IDs only, no bytes). Applied to
listPackages, listDisplayItems, searchPackages, and listUngroupedPackages.

Verified: same concurrent load that drove memory to 508/512MB and forced
a restart now peaks at 156MB (30%) with zero failed requests and no restart.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-22 19:05:34 +02:00
3b7202a662 harden(telegram): reject non-positive topic id in disableActiveTopic
All checks were successful
continuous-integration/drone/push Build is passing
BigInt("") / BigInt("0") don't throw; guard against them since the
action is exported and admin-callable independent of the UI button.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-22 12:15:03 +02:00
23f8e91c50 feat(telegram): add "skip & disable topic" on the live worker panel
Track the forum topic currently being processed on IngestionRun
(currentTopicId + currentAccountChannelMapId; additive migration) and
expose it on the live status. processArchiveSets gains an optional
shouldStop callback polled before each archive set; the forum branch
passes a live isTopicFetchEnabled check, so disabling a topic mid-run
lets the in-flight file finish, then skips the rest of that topic.

A new disableActiveTopic server action sets the topic's fetchEnabled
false (upsert), and the worker status panel shows a "Skip & disable
this topic" button while a topic is being processed. Future runs skip
the topic via the existing live per-topic read.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-22 12:10:56 +02:00
14 changed files with 1045 additions and 18 deletions

View File

@@ -0,0 +1,454 @@
# Send All From Creator Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Add a toolbar button on the STL packages page that queues every sendable package from the currently-filtered creator for delivery to the user's Telegram.
**Architecture:** A new server action `sendAllFromCreatorAction(creatorName)` mirrors the existing `sendAllInGroupAction`: it fetches sendable packages for the creator, dedups against live `BotSendRequest`s, creates a `BotSendRequest` per package, and fires `pg_notify('bot_send', id)`. The bot's existing `send-listener` consumes these unchanged. The client table renders a "Send all from [Creator]" button in the packages-tab toolbar only when `?creator=<name>` is active, wired to the action with a confirm dialog and a count toast.
**Tech Stack:** Next.js 16 App Router, TypeScript, Prisma v7, server actions, `pg_notify`, sonner toasts, lucide-react icons.
**Testing note:** This repo has no test framework (`CLAUDE.md`: "Testing is manual"). Verification for each task is `npm run lint` + `npm run build`, plus manual UI checks at the end. Do not scaffold a test harness.
---
### Task 1: Add `sendAllFromCreatorAction` server action
**Files:**
- Modify: `src/app/(app)/stls/actions.ts` (append new action after `sendAllInGroupAction`, which ends at line 591)
Reference implementation to mirror: `sendAllInGroupAction` at `src/app/(app)/stls/actions.ts:515-591`.
Existing imports already present in the file (do not re-add): `auth` from `@/lib/auth`, `prisma` from `@/lib/prisma`, `ActionResult` from `@/types/api.types`, `revalidatePath` from `next/cache`.
`ActionResult` is generic: `ActionResult<T = void>` (`src/types/api.types.ts`).
- [ ] **Step 1: Append the new action**
Add this to the end of `src/app/(app)/stls/actions.ts`:
```ts
export async function sendAllFromCreatorAction(
creatorName: string
): Promise<ActionResult<{ queued: number; skipped: number }>> {
const session = await auth();
if (!session?.user?.id) return { success: false, error: "Unauthorized" };
const creator = creatorName.trim();
if (!creator) {
return { success: false, error: "No creator specified" };
}
try {
const telegramLink = await prisma.telegramLink.findUnique({
where: { userId: session.user.id },
});
if (!telegramLink) {
return { success: false, error: "No linked Telegram account. Link one in Settings." };
}
const sendablePackages = await prisma.package.findMany({
where: {
creator,
destChannelId: { not: null },
destMessageId: { not: null },
},
select: { id: true },
});
if (sendablePackages.length === 0) {
return { success: false, error: "No uploaded packages found for this creator" };
}
let queued = 0;
let skipped = 0;
for (const pkg of sendablePackages) {
// Only create if no existing PENDING/SENDING request for this package+link combo
const existing = await prisma.botSendRequest.findFirst({
where: {
packageId: pkg.id,
telegramLinkId: telegramLink.id,
status: { in: ["PENDING", "SENDING"] },
},
});
if (existing) {
skipped++;
continue;
}
const sendRequest = await prisma.botSendRequest.create({
data: {
packageId: pkg.id,
telegramLinkId: telegramLink.id,
requestedByUserId: session.user.id,
status: "PENDING",
},
});
// Notify the bot via pg_notify
try {
await prisma.$queryRawUnsafe(
`SELECT pg_notify('bot_send', $1)`,
sendRequest.id
);
} catch {
// Best-effort — the bot also polls periodically
}
queued++;
}
revalidatePath("/stls");
return { success: true, data: { queued, skipped } };
} catch {
return { success: false, error: "Failed to send creator packages" };
}
}
```
- [ ] **Step 2: Verify lint + typecheck pass**
Run: `npm run lint`
Expected: no new errors in `src/app/(app)/stls/actions.ts`.
- [ ] **Step 3: Commit**
```bash
git add src/app/\(app\)/stls/actions.ts
git commit -m "feat(stls): add sendAllFromCreatorAction server action"
```
---
### Task 2: Wire up the toolbar button in the STL table
**Files:**
- Modify: `src/app/(app)/stls/_components/stl-table.tsx`
- lucide import (line 6)
- actions import block (lines 44-54)
- handler (after `handleSendAllInGroup`, which ends at line 278)
- `activeCreator` derivation (near `activeTag` at line 445)
- toolbar JSX (packages-tab toolbar row starting at line 478)
- [ ] **Step 1: Add the `Send` icon to the lucide import**
Change line 6 from:
```ts
import { Search, Layers, Upload } from "lucide-react";
```
to:
```ts
import { Search, Layers, Upload, Send } from "lucide-react";
```
- [ ] **Step 2: Import the new action**
In the actions import block (lines 44-54), add `sendAllFromCreatorAction` next to `sendAllInGroupAction`:
```ts
import {
updatePackageCreator,
updatePackageTags,
renameGroupAction,
dissolveGroupAction,
createGroupAction,
removeFromGroupAction,
sendAllInGroupAction,
sendAllFromCreatorAction,
updateGroupPreviewAction,
mergeGroupsAction,
} from "../actions";
```
- [ ] **Step 3: Derive the active creator from the URL**
Immediately after the `activeTag` line (`src/app/(app)/stls/_components/stl-table.tsx:445`):
```ts
const activeTag = searchParams.get("tag") ?? "";
```
add:
```ts
const activeCreator = searchParams.get("creator") ?? "";
```
- [ ] **Step 4: Add the click handler**
After `handleSendAllInGroup` (which ends at line 278), add:
```ts
const handleSendAllFromCreator = useCallback(() => {
if (!confirm(`Send all packages from "${activeCreator}" to your Telegram?`)) return;
startTransition(async () => {
const result = await sendAllFromCreatorAction(activeCreator);
if (result.success) {
const { queued, skipped } = result.data;
toast.success(
`Queued ${queued} package${queued === 1 ? "" : "s"} from ${activeCreator}` +
(skipped ? ` (${skipped} already queued)` : "")
);
router.refresh();
} else {
toast.error(result.error);
}
});
}, [activeCreator, router]);
```
- [ ] **Step 5: Render the toolbar button**
In the packages-tab toolbar (`flex flex-wrap items-center gap-2` row at line 478), add the button right after the "Upload Files" button (which closes at line 507, before the `selectedPackages.size >= 2` block at line 508):
```tsx
{activeCreator && (
<Button
variant="outline"
size="sm"
className="h-9 gap-1.5"
onClick={handleSendAllFromCreator}
>
<Send className="h-3.5 w-3.5" />
Send all from {activeCreator}
</Button>
)}
```
- [ ] **Step 6: Verify lint + build pass**
Run: `npm run lint && npm run build`
Expected: no new errors; build completes.
- [ ] **Step 7: Manual verification**
1. `npm run dev`, open `/stls`.
2. Click a creator name in the Creator column to apply `?creator=<name>` (or navigate to `/stls?creator=<known creator>`).
3. Confirm the "Send all from [Creator]" button appears in the toolbar.
4. Remove the creator filter → confirm the button disappears.
5. Click the button → confirm the dialog appears; on confirm, a toast reports the queued count.
6. (If a linked Telegram account + uploaded packages exist) confirm packages arrive via the bot.
- [ ] **Step 8: Commit**
```bash
git add src/app/\(app\)/stls/_components/stl-table.tsx
git commit -m "feat(stls): add 'send all from creator' toolbar button"
```
---
### Task 3: Searchable creator filter combobox (makes the filter reachable)
**Why:** Discovered during execution — the `?creator=` filter that reveals the Task 2
button had no UI trigger. The creator table cell click only opens the edit prompt
(`package-columns.tsx:339``onSetCreator`). This task adds a searchable
"All Creators" combobox to the packages-tab toolbar (like the existing Tags select,
but type-to-filter since there can be many creators). The creator cell stays
edit-only.
**Files:**
- Modify: `src/lib/telegram/queries.ts` — add `getAllPackageCreators()`.
- Modify: `src/app/(app)/stls/page.tsx` — fetch + pass `availableCreators`.
- Create: `src/app/(app)/stls/_components/creator-filter.tsx` — the combobox.
- Modify: `src/app/(app)/stls/_components/stl-table.tsx` — new prop, handler, render.
- [ ] **Step 1: Add the distinct-creators query**
Append to `src/lib/telegram/queries.ts` (mirrors `getAllPackageTags` at line 530):
```ts
export async function getAllPackageCreators(): Promise<string[]> {
const result = await prisma.$queryRaw<{ creator: string }[]>`
SELECT DISTINCT creator FROM packages
WHERE creator IS NOT NULL AND creator <> ''
ORDER BY creator
`;
return result.map((r) => r.creator);
}
```
- [ ] **Step 2: Create the combobox component**
Create `src/app/(app)/stls/_components/creator-filter.tsx` (mirrors the
Popover+Command pattern in `src/components/shared/data-table-faceted-filter.tsx`):
```tsx
"use client";
import { useState } from "react";
import { Check, ChevronsUpDown } from "lucide-react";
import { cn } from "@/lib/utils";
import { Button } from "@/components/ui/button";
import {
Command,
CommandEmpty,
CommandGroup,
CommandInput,
CommandItem,
CommandList,
} from "@/components/ui/command";
import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover";
interface CreatorFilterProps {
creators: string[];
value: string; // active creator, "" when none
onChange: (creator: string) => void; // "" clears the filter
}
export function CreatorFilter({ creators, value, onChange }: CreatorFilterProps) {
const [open, setOpen] = useState(false);
return (
<Popover open={open} onOpenChange={setOpen}>
<PopoverTrigger asChild>
<Button
variant="outline"
size="sm"
role="combobox"
aria-expanded={open}
className="h-9 w-[200px] justify-between"
>
<span className="truncate">{value || "All Creators"}</span>
<ChevronsUpDown className="ml-2 h-4 w-4 shrink-0 opacity-50" />
</Button>
</PopoverTrigger>
<PopoverContent className="w-[240px] p-0" align="start">
<Command>
<CommandInput placeholder="Search creators..." className="h-9" />
<CommandList>
<CommandEmpty>No creators found.</CommandEmpty>
<CommandGroup>
<CommandItem
value="__all__"
onSelect={() => {
onChange("");
setOpen(false);
}}
>
<Check
className={cn("mr-2 h-4 w-4", value === "" ? "opacity-100" : "opacity-0")}
/>
All Creators
</CommandItem>
{creators.map((creator) => (
<CommandItem
key={creator}
value={creator}
onSelect={() => {
onChange(creator);
setOpen(false);
}}
>
<Check
className={cn(
"mr-2 h-4 w-4",
value === creator ? "opacity-100" : "opacity-0"
)}
/>
<span className="truncate">{creator}</span>
</CommandItem>
))}
</CommandGroup>
</CommandList>
</Command>
</PopoverContent>
</Popover>
);
}
```
- [ ] **Step 3: Wire the query through the page**
In `src/app/(app)/stls/page.tsx`:
- Add `getAllPackageCreators` to the import from `@/lib/telegram/queries` (line 3).
- Add `getAllPackageCreators()` to the `Promise.all` (line 27) and destructure
`availableCreators`:
```ts
const [result, ingestionStatus, availableTags, availableCreators, skippedCount, ungroupedCount] =
await Promise.all([
// ...existing entries unchanged...
getIngestionStatus(),
getAllPackageTags(),
getAllPackageCreators(),
countSkippedPackages(),
countUngroupedPackages(),
]);
```
(Insert `getAllPackageCreators()` immediately after `getAllPackageTags()`, and add
`availableCreators` in the matching position of the destructure.)
- Pass the prop to `<StlTable>`: `availableCreators={availableCreators}` (next to
`availableTags={availableTags}`).
- [ ] **Step 4: Add prop + handler + render in the table**
In `src/app/(app)/stls/_components/stl-table.tsx`:
- Import the component: `import { CreatorFilter } from "./creator-filter";`
- Add to `StlTableProps` (next to `availableTags: string[];`): `availableCreators: string[];`
- Add to the destructured params (next to `availableTags,`): `availableCreators,`
- Add the handler after `updateTagFilter` (ends ~line 210):
```ts
const updateCreatorFilter = useCallback(
(value: string) => {
const params = new URLSearchParams(searchParams.toString());
if (value) {
params.set("creator", value);
params.set("page", "1");
} else {
params.delete("creator");
}
router.push(`${pathname}?${params.toString()}`, { scroll: false });
},
[router, pathname, searchParams]
);
```
- Render it in the toolbar right after the Tags `Select` block (the
`{availableTags.length > 0 && (...)}` block, ~lines 507-522):
```tsx
{availableCreators.length > 0 && (
<CreatorFilter
creators={availableCreators}
value={activeCreator}
onChange={updateCreatorFilter}
/>
)}
```
- [ ] **Step 5: Verify + manual test**
Run: `npx tsc --noEmit` (filter for the touched files — expect none new). Note: full
`npm run build` fails on a pre-existing stale Prisma client issue in
`src/lib/telegram/*` unrelated to this change; do not attempt to fix it.
Manual: open `/stls`, use the "All Creators" combobox, type to filter, pick a
creator → list filters and the "Send all from [Creator]" button appears; pick
"All Creators" → filter clears and button disappears.
- [ ] **Step 6: Commit**
```bash
git add "src/lib/telegram/queries.ts" "src/app/(app)/stls/page.tsx" \
"src/app/(app)/stls/_components/creator-filter.tsx" \
"src/app/(app)/stls/_components/stl-table.tsx"
git commit -m "feat(stls): add searchable creator filter combobox to toolbar"
```
---
## Self-Review
**Spec coverage:**
- Req 1 (button when filtered by creator) → Task 2 Step 5 (`{activeCreator && ...}`).
- Req 2 (hidden when no filter) → Task 2 Step 5 conditional.
- Req 3 (confirm dialog) → Task 2 Step 4 `confirm(...)`.
- Req 4 (all pages, not current page) → Task 1 queries `prisma.package.findMany` by `creator`, unpaginated.
- Req 5 (skip not-uploaded + dedup live requests) → Task 1 `where` filter on `destChannelId`/`destMessageId` + `existing` check.
- Req 6 (report counts) → Task 1 returns `{ queued, skipped }`; Task 2 Step 4 toast.
- Req 7 (no polling) → Task 2 handler queues + `router.refresh()`, no poll loop.
- Blank-creator guard → Task 1 `creator.trim()` check.
**Placeholder scan:** No TBD/TODO/placeholder steps; all code shown in full.
**Type consistency:** Action returns `ActionResult<{ queued: number; skipped: number }>`; handler destructures `result.data.{queued,skipped}` inside the `result.success` branch (where `data` is typed). `sendAllFromCreatorAction` name matches between Task 1 definition, Task 2 import, and Task 2 call site. `activeCreator` defined once (Step 3) and used in Steps 4-5.

View File

@@ -0,0 +1,170 @@
# Send All From Creator — Design
**Date:** 2026-07-04
**Status:** Approved for planning
## Summary
Add a "Send all from [Creator]" button to the STL packages page that queues every
sendable package from a given creator for delivery to the user's Telegram. The
button appears in the packages-tab toolbar only when the list is filtered by a
single creator (`?creator=<name>`).
This reuses the existing single-send infrastructure (`BotSendRequest` +
`pg_notify('bot_send', id)` + the bot's `send-listener`) and closely mirrors the
existing `sendAllInGroupAction`. No bot, DB schema, or send-listener changes are
required.
## Context
Relevant existing pieces:
- **Creator field:** `Package.creator` is a nullable, indexed `String` on the
`Package` model (`prisma/schema.prisma:476-524`). Filtering by exact creator is
cheap.
- **Creator filter:** The packages list already supports `?creator=<name>`
(`src/app/(app)/stls/page.tsx:22,38`), rendered by the packages column as a
clickable value (`package-columns.tsx`).
- **Single send:** `send-to-telegram-button.tsx``POST /api/telegram/bot/send`
creates a `BotSendRequest` (status `PENDING`) and fires
`pg_notify('bot_send', requestId)`; the button then polls the request to
completion.
- **Existing bulk send (group):** `sendAllInGroupAction(groupId)` in
`src/app/(app)/stls/actions.ts:515-591` fetches the group's packages, filters to
those with `destChannelId` + `destMessageId`, skips any with an existing
`PENDING`/`SENDING` request for the same package + telegram link, creates a
`BotSendRequest` per remaining package, and fires `pg_notify` per package. It is
wired into the table via `handleSendAllInGroup` (`stl-table.tsx:264-278`) with a
`confirm()` dialog and a success toast — no polling.
- **Bot side:** `bot/src/send-listener.ts` processes every `BotSendRequest`
uniformly; nothing there needs to change.
## Requirements
1. When the packages tab is filtered by a single creator (`?creator=<name>` is
present), show a toolbar button labeled **"Send all from [Creator]"**.
2. When no creator filter is active, the button is not rendered.
3. Clicking the button shows a confirmation dialog before sending (consistent with
the single-send and group-send flows, since this pushes files to Telegram).
4. On confirm, queue **all** sendable packages from that creator across all pages —
not just the current page.
5. Skip packages that are not yet uploaded (missing `destChannelId` /
`destMessageId`) and packages that already have a `PENDING`/`SENDING` request for
the same package + telegram link (dedup).
6. Report real counts back to the user via toast, e.g.
"Queued 12 packages from [Creator]" and, when applicable, a skipped count.
7. No polling — the bulk action queues and returns; sends process in the background
via the bot listener (same behavior as group send).
## Design
### Server action — `sendAllFromCreatorAction(creatorName: string)`
New action in `src/app/(app)/stls/actions.ts`, mirroring `sendAllInGroupAction`:
1. Auth check (`auth()`), return `Unauthorized` if no session.
2. Load the user's `TelegramLink`; if none, return
"No linked Telegram account. Link one in Settings."
3. Reject an empty/blank `creatorName` (guard against sending the entire "no
creator" set — see Edge cases).
4. Fetch sendable packages directly with a `where` filter (no in-memory filtering):
```ts
prisma.package.findMany({
where: {
creator: creatorName,
destChannelId: { not: null },
destMessageId: { not: null },
},
select: { id: true },
})
```
5. If none, return "No uploaded packages found for this creator."
6. For each package, skip if an existing `PENDING`/`SENDING` `BotSendRequest`
exists for that package + telegram link; otherwise create a `BotSendRequest`
(status `PENDING`) and fire `pg_notify('bot_send', id)` (best-effort, wrapped in
try/catch — the bot also polls).
7. `revalidatePath("/stls")`.
8. Return counts so the UI can show them. This is a small improvement over
`sendAllInGroupAction`, which currently returns `{ success: true, data: undefined }`.
Shape:
```ts
{ success: true, data: { queued: number, skipped: number } }
```
where `skipped` counts packages that already had a live request. (Packages not
yet uploaded are excluded by the query and not counted as skipped.)
### UI — toolbar button in `stl-table.tsx`
- Derive the active creator from the URL: `const activeCreator =
searchParams.get("creator") ?? "";` (parallel to the existing
`activeTag` at `stl-table.tsx:445`).
- In the packages-tab toolbar (`stl-table.tsx:478` `flex flex-wrap items-center
gap-2` row), render the button only when `activeCreator` is non-empty:
```tsx
{activeCreator && (
<Button variant="outline" size="sm" className="h-9 gap-1.5"
onClick={handleSendAllFromCreator}>
<Send className="h-3.5 w-3.5" />
Send all from {activeCreator}
</Button>
)}
```
(Reuse the icon already used by the send action; place near the Upload / Group
buttons.)
- Add `handleSendAllFromCreator`, modeled on `handleSendAllInGroup`
(`stl-table.tsx:264-278`):
```ts
const handleSendAllFromCreator = useCallback(() => {
if (!confirm(`Send all packages from "${activeCreator}" to your Telegram?`)) return;
startTransition(async () => {
const result = await sendAllFromCreatorAction(activeCreator);
if (result.success) {
const { queued, skipped } = result.data;
toast.success(
`Queued ${queued} package${queued === 1 ? "" : "s"} from ${activeCreator}` +
(skipped ? ` (${skipped} already queued)` : "")
);
router.refresh();
} else {
toast.error(result.error);
}
});
}, [activeCreator, router]);
```
- Import `sendAllFromCreatorAction` alongside the existing `sendAllInGroupAction`
import (`stl-table.tsx:51`).
### Data flow
```
User (filtered by creator) → clicks button → confirm dialog
→ sendAllFromCreatorAction(creatorName)
→ for each sendable, un-queued package: create BotSendRequest + pg_notify
→ returns { queued, skipped } → toast + router.refresh()
(background) bot send-listener consumes each BotSendRequest → delivers to user
```
## Edge cases
- **Blank creator:** The action rejects an empty/whitespace `creatorName` so a
stray `?creator=` never queues every package. The UI already gates the button on
a non-empty `activeCreator`, so this is defense-in-depth.
- **No linked Telegram account:** Returns an error toast, queues nothing.
- **No uploaded packages for creator:** Returns an error toast, queues nothing.
- **All packages already queued:** `queued: 0`, `skipped: N` → toast reflects it.
- **pg_notify failure:** Swallowed per-package (best-effort); the bot's periodic
poll picks the request up. Matches existing behavior.
## Out of scope / non-goals
- No per-row "send all from this creator" action (toolbar-only, per decision).
- No progress polling / live completion tracking for the bulk send.
- No changes to the bot, `send-listener`, `BotSendRequest` schema, or API routes.
- No batching/throttling beyond what the bot listener already does.
## Files touched
- `src/app/(app)/stls/actions.ts` — add `sendAllFromCreatorAction`.
- `src/app/(app)/stls/_components/stl-table.tsx` — derive `activeCreator`, add
`handleSendAllFromCreator`, render the toolbar button, import the new action.

View File

@@ -0,0 +1,6 @@
-- AlterTable: track the forum topic currently being processed on the live run
-- so the worker status panel can offer a "skip & disable topic" action.
-- Additive, nullable — no data change for existing rows.
ALTER TABLE "ingestion_runs"
ADD COLUMN "currentTopicId" BIGINT,
ADD COLUMN "currentAccountChannelMapId" TEXT;

View File

@@ -582,6 +582,8 @@ model IngestionRun {
totalBytes BigInt? // Total size of current download totalBytes BigInt? // Total size of current download
downloadPercent Int? // 0-100 downloadPercent Int? // 0-100
lastActivityAt DateTime? // When activity was last updated lastActivityAt DateTime? // When activity was last updated
currentTopicId BigInt? // Forum topic currently being processed (live "skip topic")
currentAccountChannelMapId String? // AccountChannelMap owning the topic being processed
account TelegramAccount @relation(fields: [accountId], references: [id]) account TelegramAccount @relation(fields: [accountId], references: [id])
packages Package[] packages Package[]

View File

@@ -0,0 +1,82 @@
"use client";
import { useState } from "react";
import { Check, ChevronsUpDown } from "lucide-react";
import { cn } from "@/lib/utils";
import { Button } from "@/components/ui/button";
import {
Command,
CommandEmpty,
CommandGroup,
CommandInput,
CommandItem,
CommandList,
} from "@/components/ui/command";
import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover";
interface CreatorFilterProps {
creators: string[];
value: string; // active creator, "" when none
onChange: (creator: string) => void; // "" clears the filter
}
export function CreatorFilter({ creators, value, onChange }: CreatorFilterProps) {
const [open, setOpen] = useState(false);
return (
<Popover open={open} onOpenChange={setOpen}>
<PopoverTrigger asChild>
<Button
variant="outline"
size="sm"
role="combobox"
aria-expanded={open}
className="h-9 w-[200px] justify-between"
>
<span className="truncate">{value || "All Creators"}</span>
<ChevronsUpDown className="ml-2 h-4 w-4 shrink-0 opacity-50" />
</Button>
</PopoverTrigger>
<PopoverContent className="w-[240px] p-0" align="start">
<Command>
<CommandInput placeholder="Search creators..." className="h-9" />
<CommandList>
<CommandEmpty>No creators found.</CommandEmpty>
<CommandGroup>
<CommandItem
value="__all__"
onSelect={() => {
onChange("");
setOpen(false);
}}
>
<Check
className={cn("mr-2 h-4 w-4", value === "" ? "opacity-100" : "opacity-0")}
/>
All Creators
</CommandItem>
{creators.map((creator) => (
<CommandItem
key={creator}
value={creator}
onSelect={() => {
onChange(creator);
setOpen(false);
}}
>
<Check
className={cn(
"mr-2 h-4 w-4",
value === creator ? "opacity-100" : "opacity-0"
)}
/>
<span className="truncate">{creator}</span>
</CommandItem>
))}
</CommandGroup>
</CommandList>
</Command>
</PopoverContent>
</Popover>
);
}

View File

@@ -3,7 +3,7 @@
import { useState, useCallback, useTransition, useMemo, useRef } from "react"; import { useState, useCallback, useTransition, useMemo, useRef } from "react";
import { useRouter, usePathname, useSearchParams } from "next/navigation"; import { useRouter, usePathname, useSearchParams } from "next/navigation";
import { toast } from "sonner"; import { toast } from "sonner";
import { Search, Layers, Upload } from "lucide-react"; import { Search, Layers, Upload, Send } from "lucide-react";
import { UploadDialog } from "./upload-dialog"; import { UploadDialog } from "./upload-dialog";
import { useDataTable } from "@/hooks/use-data-table"; import { useDataTable } from "@/hooks/use-data-table";
import { import {
@@ -16,6 +16,7 @@ import {
import { PackageFilesDrawer } from "./package-files-drawer"; import { PackageFilesDrawer } from "./package-files-drawer";
import { IngestionStatus } from "./ingestion-status"; import { IngestionStatus } from "./ingestion-status";
import { SkippedPackagesTab } from "./skipped-packages-tab"; import { SkippedPackagesTab } from "./skipped-packages-tab";
import { CreatorFilter } from "./creator-filter";
import { DataTable } from "@/components/shared/data-table"; import { DataTable } from "@/components/shared/data-table";
import { DataTablePagination } from "@/components/shared/data-table-pagination"; import { DataTablePagination } from "@/components/shared/data-table-pagination";
import { DataTableViewOptions } from "@/components/shared/data-table-view-options"; import { DataTableViewOptions } from "@/components/shared/data-table-view-options";
@@ -49,6 +50,7 @@ import {
createGroupAction, createGroupAction,
removeFromGroupAction, removeFromGroupAction,
sendAllInGroupAction, sendAllInGroupAction,
sendAllFromCreatorAction,
updateGroupPreviewAction, updateGroupPreviewAction,
mergeGroupsAction, mergeGroupsAction,
} from "../actions"; } from "../actions";
@@ -59,6 +61,7 @@ interface StlTableProps {
totalCount: number; totalCount: number;
ingestionStatus: IngestionAccountStatus[]; ingestionStatus: IngestionAccountStatus[];
availableTags: string[]; availableTags: string[];
availableCreators: string[];
searchTerm: string; searchTerm: string;
skippedData: SkippedRow[]; skippedData: SkippedRow[];
skippedPageCount: number; skippedPageCount: number;
@@ -74,6 +77,7 @@ export function StlTable({
totalCount, totalCount,
ingestionStatus, ingestionStatus,
availableTags, availableTags,
availableCreators,
searchTerm, searchTerm,
skippedData, skippedData,
skippedPageCount, skippedPageCount,
@@ -85,6 +89,7 @@ export function StlTable({
const router = useRouter(); const router = useRouter();
const pathname = usePathname(); const pathname = usePathname();
const searchParams = useSearchParams(); const searchParams = useSearchParams();
const activeCreator = searchParams.get("creator") ?? "";
const [searchValue, setSearchValue] = useState(searchParams.get("search") ?? ""); const [searchValue, setSearchValue] = useState(searchParams.get("search") ?? "");
const [viewPkg, setViewPkg] = useState<PackageRow | null>(null); const [viewPkg, setViewPkg] = useState<PackageRow | null>(null);
@@ -207,6 +212,20 @@ export function StlTable({
[router, pathname, searchParams] [router, pathname, searchParams]
); );
const updateCreatorFilter = useCallback(
(value: string) => {
const params = new URLSearchParams(searchParams.toString());
if (value) {
params.set("creator", value);
params.set("page", "1");
} else {
params.delete("creator");
}
router.push(`${pathname}?${params.toString()}`, { scroll: false });
},
[router, pathname, searchParams]
);
const activeTab = searchParams.get("tab") ?? "packages"; const activeTab = searchParams.get("tab") ?? "packages";
const updateTab = useCallback( const updateTab = useCallback(
@@ -277,6 +296,23 @@ export function StlTable({
[router] [router]
); );
const handleSendAllFromCreator = useCallback(() => {
if (!confirm(`Send all packages from "${activeCreator}" to your Telegram?`)) return;
startTransition(async () => {
const result = await sendAllFromCreatorAction(activeCreator);
if (result.success) {
const { queued, skipped } = result.data;
toast.success(
`Queued ${queued} package${queued === 1 ? "" : "s"} from ${activeCreator}` +
(skipped ? ` (${skipped} already queued)` : "")
);
router.refresh();
} else {
toast.error(result.error);
}
});
}, [activeCreator, router]);
const handleRemoveFromGroup = useCallback( const handleRemoveFromGroup = useCallback(
(packageId: string) => { (packageId: string) => {
startTransition(async () => { startTransition(async () => {
@@ -500,11 +536,29 @@ export function StlTable({
</SelectContent> </SelectContent>
</Select> </Select>
)} )}
{availableCreators.length > 0 && (
<CreatorFilter
creators={availableCreators}
value={activeCreator}
onChange={updateCreatorFilter}
/>
)}
<DataTableViewOptions table={table} /> <DataTableViewOptions table={table} />
<Button variant="outline" size="sm" className="h-9" onClick={() => setUploadOpen(true)}> <Button variant="outline" size="sm" className="h-9" onClick={() => setUploadOpen(true)}>
<Upload className="mr-2 h-4 w-4" /> <Upload className="mr-2 h-4 w-4" />
Upload Files Upload Files
</Button> </Button>
{activeCreator && (
<Button
variant="outline"
size="sm"
className="h-9 gap-1.5"
onClick={handleSendAllFromCreator}
>
<Send className="h-3.5 w-3.5" />
Send all from {activeCreator}
</Button>
)}
{selectedPackages.size >= 2 && ( {selectedPackages.size >= 2 && (
<Button <Button
variant="outline" variant="outline"

View File

@@ -589,3 +589,82 @@ export async function sendAllInGroupAction(
return { success: false, error: "Failed to send group packages" }; return { success: false, error: "Failed to send group packages" };
} }
} }
export async function sendAllFromCreatorAction(
creatorName: string
): Promise<ActionResult<{ queued: number; skipped: number }>> {
const session = await auth();
if (!session?.user?.id) return { success: false, error: "Unauthorized" };
const creator = creatorName.trim();
if (!creator) {
return { success: false, error: "No creator specified" };
}
try {
const telegramLink = await prisma.telegramLink.findUnique({
where: { userId: session.user.id },
});
if (!telegramLink) {
return { success: false, error: "No linked Telegram account. Link one in Settings." };
}
const sendablePackages = await prisma.package.findMany({
where: {
creator,
destChannelId: { not: null },
destMessageId: { not: null },
},
select: { id: true },
});
if (sendablePackages.length === 0) {
return { success: false, error: "No uploaded packages found for this creator" };
}
let queued = 0;
let skipped = 0;
for (const pkg of sendablePackages) {
// Only create if no existing PENDING/SENDING request for this package+link combo
const existing = await prisma.botSendRequest.findFirst({
where: {
packageId: pkg.id,
telegramLinkId: telegramLink.id,
status: { in: ["PENDING", "SENDING"] },
},
});
if (existing) {
skipped++;
continue;
}
const sendRequest = await prisma.botSendRequest.create({
data: {
packageId: pkg.id,
telegramLinkId: telegramLink.id,
requestedByUserId: session.user.id,
status: "PENDING",
},
});
// Notify the bot via pg_notify
try {
await prisma.$queryRawUnsafe(
`SELECT pg_notify('bot_send', $1)`,
sendRequest.id
);
} catch {
// Best-effort — the bot also polls periodically
}
queued++;
}
revalidatePath("/stls");
return { success: true, data: { queued, skipped } };
} catch {
return { success: false, error: "Failed to send creator packages" };
}
}

View File

@@ -1,6 +1,6 @@
import { auth } from "@/lib/auth"; import { auth } from "@/lib/auth";
import { redirect } from "next/navigation"; import { redirect } from "next/navigation";
import { listDisplayItems, searchPackages, getIngestionStatus, getAllPackageTags, listSkippedPackages, countSkippedPackages, listUngroupedPackages, countUngroupedPackages } from "@/lib/telegram/queries"; import { listDisplayItems, searchPackages, getIngestionStatus, getAllPackageTags, getAllPackageCreators, listSkippedPackages, countSkippedPackages, listUngroupedPackages, countUngroupedPackages } from "@/lib/telegram/queries";
import { StlTable } from "./_components/stl-table"; import { StlTable } from "./_components/stl-table";
import type { DisplayItem, PackageListItem } from "@/lib/telegram/types"; import type { DisplayItem, PackageListItem } from "@/lib/telegram/types";
@@ -24,7 +24,7 @@ export default async function StlFilesPage({ searchParams }: Props) {
const tab = (params.tab as string) ?? "packages"; const tab = (params.tab as string) ?? "packages";
// Fetch packages, ingestion status, tags, and skipped count in parallel // Fetch packages, ingestion status, tags, and skipped count in parallel
const [result, ingestionStatus, availableTags, skippedCount, ungroupedCount] = await Promise.all([ const [result, ingestionStatus, availableTags, availableCreators, skippedCount, ungroupedCount] = await Promise.all([
search search
? searchPackages({ ? searchPackages({
query: search, query: search,
@@ -42,6 +42,7 @@ export default async function StlFilesPage({ searchParams }: Props) {
}), }),
getIngestionStatus(), getIngestionStatus(),
getAllPackageTags(), getAllPackageTags(),
getAllPackageCreators(),
countSkippedPackages(), countSkippedPackages(),
countUngroupedPackages(), countUngroupedPackages(),
]); ]);
@@ -68,6 +69,7 @@ export default async function StlFilesPage({ searchParams }: Props) {
totalCount={result.pagination.total} totalCount={result.pagination.total}
ingestionStatus={ingestionStatus} ingestionStatus={ingestionStatus}
availableTags={availableTags} availableTags={availableTags}
availableCreators={availableCreators}
searchTerm={search} searchTerm={search}
skippedData={skippedResult?.items ?? []} skippedData={skippedResult?.items ?? []}
skippedPageCount={skippedResult?.pagination.totalPages ?? 0} skippedPageCount={skippedResult?.pagination.totalPages ?? 0}

View File

@@ -9,13 +9,14 @@ import {
Radio, Radio,
AlertTriangle, AlertTriangle,
RefreshCw, RefreshCw,
SkipForward,
} from "lucide-react"; } from "lucide-react";
import { Card, CardContent } from "@/components/ui/card"; import { Card, CardContent } from "@/components/ui/card";
import { Badge } from "@/components/ui/badge"; import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button"; import { Button } from "@/components/ui/button";
import { cn } from "@/lib/utils"; import { cn } from "@/lib/utils";
import { toast } from "sonner"; import { toast } from "sonner";
import { triggerIngestion } from "../actions"; import { triggerIngestion, disableActiveTopic } from "../actions";
import type { IngestionAccountStatus } from "@/lib/telegram/types"; import type { IngestionAccountStatus } from "@/lib/telegram/types";
interface WorkerStatusPanelProps { interface WorkerStatusPanelProps {
@@ -218,6 +219,24 @@ function RunningStatus({
}: { }: {
run: NonNullable<IngestionAccountStatus["currentRun"]>; run: NonNullable<IngestionAccountStatus["currentRun"]>;
}) { }) {
const [isDisabling, startDisable] = useTransition();
const handleSkipTopic = () => {
const acmId = run.currentAccountChannelMapId;
const topicId = run.currentTopicId;
if (!acmId || !topicId) return;
startDisable(async () => {
const result = await disableActiveTopic(acmId, topicId);
if (result.success) {
toast.success(
"Topic disabled — the current file finishes, then the rest is skipped"
);
} else {
toast.error(result.error ?? "Failed to disable topic");
}
});
};
return ( return (
<div className="space-y-2"> <div className="space-y-2">
<div className="flex items-center gap-2"> <div className="flex items-center gap-2">
@@ -273,6 +292,27 @@ function RunningStatus({
</span> </span>
)} )}
</div> </div>
{/* Skip & disable the topic currently being processed */}
{run.currentTopicId && run.currentAccountChannelMapId && (
<div className="pl-6 pt-1">
<Button
variant="outline"
size="sm"
className="h-7 gap-1.5 text-xs"
onClick={handleSkipTopic}
disabled={isDisabling}
title="Finish the current file, skip the rest of this topic, and don't fetch it in future runs"
>
{isDisabling ? (
<Loader2 className="h-3 w-3 animate-spin" />
) : (
<SkipForward className="h-3 w-3" />
)}
Skip &amp; disable this topic
</Button>
</div>
)}
</div> </div>
); );
} }

View File

@@ -462,6 +462,54 @@ export async function setTopicFetchEnabled(
} }
} }
/**
* Disable the topic currently being processed by the worker, identified by its
* account-channel mapping + Telegram topic id (as exposed on the live run
* status). Upserts so a disabled row exists even if the worker hasn't persisted
* the topic yet. The worker honours this live: it finishes the in-flight file
* then skips the rest of the topic, and future runs skip it entirely.
*/
export async function disableActiveTopic(
accountChannelMapId: string,
topicId: string
): Promise<ActionResult> {
const admin = await requireAdmin();
if (!admin.success) return admin;
let topicIdBig: bigint;
try {
topicIdBig = BigInt(topicId);
} catch {
return { success: false, error: "Invalid topic id" };
}
// BigInt("") / BigInt("0") don't throw — reject non-positive ids explicitly
// since this action is exported and admin-callable outside the UI button.
if (topicIdBig <= BigInt(0)) {
return { success: false, error: "Invalid topic id" };
}
try {
await prisma.topicProgress.upsert({
where: {
accountChannelMapId_topicId: {
accountChannelMapId,
topicId: topicIdBig,
},
},
create: {
accountChannelMapId,
topicId: topicIdBig,
fetchEnabled: false,
},
update: { fetchEnabled: false },
});
revalidatePath(REVALIDATE_PATH);
return { success: true, data: undefined };
} catch {
return { success: false, error: "Failed to disable topic" };
}
}
// ── Account-Channel link actions ── // ── Account-Channel link actions ──
export async function linkChannel( export async function linkChannel(

View File

@@ -9,6 +9,28 @@ import type {
PackageGroupRow, PackageGroupRow,
} from "./types"; } from "./types";
/**
* Returns the subset of the given IDs whose `previewData` column is non-null,
* WITHOUT transferring the (large) image bytes.
*
* List views only need a `hasPreview` boolean per row. Selecting `previewData`
* directly pulls every JPEG blob (avg ~700 KB, up to 2 MB) into the Node heap
* just to compare it against null — under concurrent requests this exhausts the
* container memory limit and crashes the process. This keeps the check in SQL.
*/
async function fetchPreviewFlags(
table: "packages" | "package_groups",
ids: string[]
): Promise<Set<string>> {
if (ids.length === 0) return new Set();
const placeholders = ids.map((_, i) => `$${i + 1}`).join(", ");
const rows = await prisma.$queryRawUnsafe<{ id: string }[]>(
`SELECT id FROM ${table} WHERE "previewData" IS NOT NULL AND id IN (${placeholders})`,
...ids
);
return new Set(rows.map((r) => r.id));
}
export async function listPackages(options: { export async function listPackages(options: {
page: number; page: number;
limit: number; limit: number;
@@ -40,13 +62,17 @@ export async function listPackages(options: {
indexedAt: true, indexedAt: true,
creator: true, creator: true,
tags: true, tags: true,
previewData: true, // check actual image data, not previewMsgId proxy
sourceChannel: { select: { id: true, title: true } }, sourceChannel: { select: { id: true, title: true } },
}, },
}), }),
prisma.package.count({ where }), prisma.package.count({ where }),
]); ]);
const previewIds = await fetchPreviewFlags(
"packages",
items.map((p) => p.id)
);
const mapped: PackageListItem[] = items.map((pkg) => ({ const mapped: PackageListItem[] = items.map((pkg) => ({
id: pkg.id, id: pkg.id,
fileName: pkg.fileName, fileName: pkg.fileName,
@@ -55,7 +81,7 @@ export async function listPackages(options: {
archiveType: pkg.archiveType, archiveType: pkg.archiveType,
fileCount: pkg.fileCount, fileCount: pkg.fileCount,
isMultipart: pkg.isMultipart, isMultipart: pkg.isMultipart,
hasPreview: pkg.previewData !== null, hasPreview: previewIds.has(pkg.id),
creator: pkg.creator, creator: pkg.creator,
tags: pkg.tags, tags: pkg.tags,
indexedAt: pkg.indexedAt.toISOString(), indexedAt: pkg.indexedAt.toISOString(),
@@ -149,7 +175,7 @@ export async function listDisplayItems(options: {
select: { select: {
id: true, fileName: true, fileSize: true, contentHash: true, id: true, fileName: true, fileSize: true, contentHash: true,
archiveType: true, fileCount: true, isMultipart: true, archiveType: true, fileCount: true, isMultipart: true,
indexedAt: true, creator: true, tags: true, previewData: true, indexedAt: true, creator: true, tags: true,
sourceChannel: { select: { id: true, title: true } }, sourceChannel: { select: { id: true, title: true } },
}, },
}) })
@@ -159,13 +185,13 @@ export async function listDisplayItems(options: {
? await prisma.packageGroup.findMany({ ? await prisma.packageGroup.findMany({
where: { id: { in: groupIds } }, where: { id: { in: groupIds } },
select: { select: {
id: true, name: true, previewData: true, id: true, name: true,
sourceChannel: { select: { id: true, title: true } }, sourceChannel: { select: { id: true, title: true } },
packages: { packages: {
select: { select: {
id: true, fileName: true, fileSize: true, contentHash: true, id: true, fileName: true, fileSize: true, contentHash: true,
archiveType: true, fileCount: true, isMultipart: true, archiveType: true, fileCount: true, isMultipart: true,
indexedAt: true, creator: true, tags: true, previewData: true, indexedAt: true, creator: true, tags: true,
sourceChannel: { select: { id: true, title: true } }, sourceChannel: { select: { id: true, title: true } },
}, },
orderBy: { indexedAt: "desc" }, orderBy: { indexedAt: "desc" },
@@ -174,6 +200,16 @@ export async function listDisplayItems(options: {
}) })
: []; : [];
// Compute hasPreview flags without transferring image bytes (see fetchPreviewFlags)
const allPackageIds = [
...standalonePackages.map((p) => p.id),
...groups.flatMap((g) => g.packages.map((p) => p.id)),
];
const [packagePreviewIds, groupPreviewIds] = await Promise.all([
fetchPreviewFlags("packages", allPackageIds),
fetchPreviewFlags("package_groups", groups.map((g) => g.id)),
]);
// Build DisplayItem array in the original sort order // Build DisplayItem array in the original sort order
const packageMap = new Map(standalonePackages.map((p) => [p.id, p])); const packageMap = new Map(standalonePackages.map((p) => [p.id, p]));
const groupMap = new Map(groups.map((g) => [g.id, g])); const groupMap = new Map(groups.map((g) => [g.id, g]));
@@ -191,7 +227,7 @@ export async function listDisplayItems(options: {
archiveType: pkg.archiveType, archiveType: pkg.archiveType,
fileCount: pkg.fileCount, fileCount: pkg.fileCount,
isMultipart: pkg.isMultipart, isMultipart: pkg.isMultipart,
hasPreview: pkg.previewData !== null, hasPreview: packagePreviewIds.has(pkg.id),
creator: pkg.creator, creator: pkg.creator,
tags: pkg.tags, tags: pkg.tags,
indexedAt: pkg.indexedAt.toISOString(), indexedAt: pkg.indexedAt.toISOString(),
@@ -209,7 +245,7 @@ export async function listDisplayItems(options: {
data: { data: {
id: grp.id, id: grp.id,
name: grp.name, name: grp.name,
hasPreview: grp.previewData !== null, hasPreview: groupPreviewIds.has(grp.id),
totalFileSize: grp.packages.reduce((sum, p) => sum + p.fileSize, BigInt(0)).toString(), totalFileSize: grp.packages.reduce((sum, p) => sum + p.fileSize, BigInt(0)).toString(),
totalFileCount: grp.packages.reduce((sum, p) => sum + p.fileCount, 0), totalFileCount: grp.packages.reduce((sum, p) => sum + p.fileCount, 0),
packageCount: grp.packages.length, packageCount: grp.packages.length,
@@ -227,7 +263,7 @@ export async function listDisplayItems(options: {
archiveType: pkg.archiveType, archiveType: pkg.archiveType,
fileCount: pkg.fileCount, fileCount: pkg.fileCount,
isMultipart: pkg.isMultipart, isMultipart: pkg.isMultipart,
hasPreview: pkg.previewData !== null, hasPreview: packagePreviewIds.has(pkg.id),
creator: pkg.creator, creator: pkg.creator,
tags: pkg.tags, tags: pkg.tags,
indexedAt: pkg.indexedAt.toISOString(), indexedAt: pkg.indexedAt.toISOString(),
@@ -440,13 +476,17 @@ export async function searchPackages(options: {
indexedAt: true, indexedAt: true,
creator: true, creator: true,
tags: true, tags: true,
previewData: true,
sourceChannel: { select: { id: true, title: true } }, sourceChannel: { select: { id: true, title: true } },
}, },
}), }),
Promise.resolve(allIds.length), Promise.resolve(allIds.length),
]); ]);
const previewIds = await fetchPreviewFlags(
"packages",
items.map((p) => p.id)
);
const mapped: PackageListItem[] = items.map((pkg) => ({ const mapped: PackageListItem[] = items.map((pkg) => ({
id: pkg.id, id: pkg.id,
fileName: pkg.fileName, fileName: pkg.fileName,
@@ -455,7 +495,7 @@ export async function searchPackages(options: {
archiveType: pkg.archiveType, archiveType: pkg.archiveType,
fileCount: pkg.fileCount, fileCount: pkg.fileCount,
isMultipart: pkg.isMultipart, isMultipart: pkg.isMultipart,
hasPreview: pkg.previewData !== null, hasPreview: previewIds.has(pkg.id),
creator: pkg.creator, creator: pkg.creator,
tags: pkg.tags, tags: pkg.tags,
indexedAt: pkg.indexedAt.toISOString(), indexedAt: pkg.indexedAt.toISOString(),
@@ -494,6 +534,15 @@ export async function getAllPackageTags(): Promise<string[]> {
return result.map((r) => r.tag); return result.map((r) => r.tag);
} }
export async function getAllPackageCreators(): Promise<string[]> {
const result = await prisma.$queryRaw<{ creator: string }[]>`
SELECT DISTINCT creator FROM packages
WHERE creator IS NOT NULL AND creator <> ''
ORDER BY creator
`;
return result.map((r) => r.creator);
}
export async function getIngestionStatus(): Promise<IngestionAccountStatus[]> { export async function getIngestionStatus(): Promise<IngestionAccountStatus[]> {
const accounts = await prisma.telegramAccount.findMany({ const accounts = await prisma.telegramAccount.findMany({
orderBy: { createdAt: "asc" }, orderBy: { createdAt: "asc" },
@@ -550,6 +599,8 @@ export async function getIngestionStatus(): Promise<IngestionAccountStatus[]> {
totalBytes: currentRun.totalBytes?.toString() ?? null, totalBytes: currentRun.totalBytes?.toString() ?? null,
downloadPercent: currentRun.downloadPercent, downloadPercent: currentRun.downloadPercent,
lastActivityAt: currentRun.lastActivityAt?.toISOString() ?? null, lastActivityAt: currentRun.lastActivityAt?.toISOString() ?? null,
currentTopicId: currentRun.currentTopicId?.toString() ?? null,
currentAccountChannelMapId: currentRun.currentAccountChannelMapId,
} }
: null, : null,
}); });
@@ -634,13 +685,17 @@ export async function listUngroupedPackages(options: {
partCount: true, partCount: true,
tags: true, tags: true,
indexedAt: true, indexedAt: true,
previewData: true,
sourceChannel: { select: { id: true, title: true } }, sourceChannel: { select: { id: true, title: true } },
}, },
}), }),
prisma.package.count({ where }), prisma.package.count({ where }),
]); ]);
const previewIds = await fetchPreviewFlags(
"packages",
items.map((p) => p.id)
);
return { return {
items: items.map((p) => ({ items: items.map((p) => ({
id: p.id, id: p.id,
@@ -654,7 +709,7 @@ export async function listUngroupedPackages(options: {
partCount: p.partCount, partCount: p.partCount,
tags: p.tags, tags: p.tags,
indexedAt: p.indexedAt.toISOString(), indexedAt: p.indexedAt.toISOString(),
hasPreview: !!p.previewData, hasPreview: previewIds.has(p.id),
sourceChannel: p.sourceChannel, sourceChannel: p.sourceChannel,
matchedFileCount: 0, matchedFileCount: 0,
matchedByContent: false, matchedByContent: false,

View File

@@ -123,5 +123,7 @@ export interface IngestionAccountStatus {
totalBytes: string | null; // BigInt serialized as string totalBytes: string | null; // BigInt serialized as string
downloadPercent: number | null; downloadPercent: number | null;
lastActivityAt: string | null; lastActivityAt: string | null;
currentTopicId: string | null; // BigInt serialized as string
currentAccountChannelMapId: string | null;
} | null; } | null;
} }

View File

@@ -369,6 +369,8 @@ export interface ActivityUpdate {
downloadedBytes?: bigint | null; downloadedBytes?: bigint | null;
totalBytes?: bigint | null; totalBytes?: bigint | null;
downloadPercent?: number | null; downloadPercent?: number | null;
currentTopicId?: bigint | null;
currentAccountChannelMapId?: string | null;
messagesScanned?: number; messagesScanned?: number;
zipsFound?: number; zipsFound?: number;
zipsDuplicate?: number; zipsDuplicate?: number;
@@ -396,6 +398,10 @@ export async function updateRunActivity(
...(activity.zipsFound !== undefined && { zipsFound: activity.zipsFound }), ...(activity.zipsFound !== undefined && { zipsFound: activity.zipsFound }),
...(activity.zipsDuplicate !== undefined && { zipsDuplicate: activity.zipsDuplicate }), ...(activity.zipsDuplicate !== undefined && { zipsDuplicate: activity.zipsDuplicate }),
...(activity.zipsIngested !== undefined && { zipsIngested: activity.zipsIngested }), ...(activity.zipsIngested !== undefined && { zipsIngested: activity.zipsIngested }),
...(activity.currentTopicId !== undefined && { currentTopicId: activity.currentTopicId }),
...(activity.currentAccountChannelMapId !== undefined && {
currentAccountChannelMapId: activity.currentAccountChannelMapId,
}),
}, },
}); });
} }
@@ -410,6 +416,8 @@ const CLEAR_ACTIVITY = {
downloadedBytes: null, downloadedBytes: null,
totalBytes: null, totalBytes: null,
downloadPercent: null, downloadPercent: null,
currentTopicId: null,
currentAccountChannelMapId: null,
lastActivityAt: new Date(), lastActivityAt: new Date(),
}; };

View File

@@ -538,6 +538,8 @@ export async function runWorkerForAccount(
currentActivity: `Enumerating topics in "${channelLabel}"`, currentActivity: `Enumerating topics in "${channelLabel}"`,
currentStep: "scanning", currentStep: "scanning",
currentChannel: channelLabel, currentChannel: channelLabel,
currentTopicId: null,
currentAccountChannelMapId: null,
currentFile: null, currentFile: null,
currentFileNum: null, currentFileNum: null,
totalFiles: null, totalFiles: null,
@@ -750,6 +752,8 @@ export async function runWorkerForAccount(
currentActivity: `Scanning "${topicLabel}"${topicProgress}`, currentActivity: `Scanning "${topicLabel}"${topicProgress}`,
currentStep: "scanning", currentStep: "scanning",
currentChannel: channelLabel, currentChannel: channelLabel,
currentTopicId: topic.topicId,
currentAccountChannelMapId: mapping.id,
currentFile: null, currentFile: null,
currentFileNum: null, currentFileNum: null,
totalFiles: null, totalFiles: null,
@@ -830,7 +834,11 @@ export async function runWorkerForAccount(
topic.name, topic.name,
messageId messageId
); );
} },
// shouldStop: re-read the live fetch flag before each archive
// set. A mid-run "disable topic" lets the current file finish,
// then skips the rest of this topic's archives.
async () => !(await isTopicFetchEnabled(mapping.id, topic.topicId))
); );
// Sync client back in case it was recreated during upload stall recovery // Sync client back in case it was recreated during upload stall recovery
client = pipelineCtx.client; client = pipelineCtx.client;
@@ -922,6 +930,8 @@ export async function runWorkerForAccount(
currentActivity: `Scanning "${channelLabel}" for new archives`, currentActivity: `Scanning "${channelLabel}" for new archives`,
currentStep: "scanning", currentStep: "scanning",
currentChannel: channelLabel, currentChannel: channelLabel,
currentTopicId: null,
currentAccountChannelMapId: null,
currentFile: null, currentFile: null,
currentFileNum: null, currentFileNum: null,
totalFiles: null, totalFiles: null,
@@ -1203,7 +1213,12 @@ async function processArchiveSets(
* below any failed message ID in this scan). Used by the caller to * below any failed message ID in this scan). Used by the caller to
* advance the channel/topic watermark incrementally — otherwise a long * advance the channel/topic watermark incrementally — otherwise a long
* scan that gets killed by worker restart loses all progress. */ * scan that gets killed by worker restart loses all progress. */
onWatermarkAdvance?: (messageId: bigint) => Promise<void> onWatermarkAdvance?: (messageId: bigint) => Promise<void>,
/** Optional cancellation check, polled before each archive set. When it
* resolves true, processing stops after the set currently in flight (that
* one completes; remaining sets in this scan are skipped). Used by the
* forum branch to honour a mid-run "disable topic". */
shouldStop?: () => Promise<boolean>
): Promise<{ maxProcessedId: bigint | null; minFailedId: bigint | null }> { ): Promise<{ maxProcessedId: bigint | null; minFailedId: bigint | null }> {
const { client, runId, channelTitle, channel, throttled, counters, accountLog } = ctx; const { client, runId, channelTitle, channel, throttled, counters, accountLog } = ctx;
@@ -1281,6 +1296,16 @@ async function processArchiveSets(
const indexedPackageRefs: IndexedPackageRef[] = []; const indexedPackageRefs: IndexedPackageRef[] = [];
for (let setIdx = 0; setIdx < archiveSets.length; setIdx++) { for (let setIdx = 0; setIdx < archiveSets.length; setIdx++) {
// Cooperative cancellation: if the caller signals stop (e.g. the topic was
// disabled mid-run), skip the remaining archive sets in this scan. The set
// processed in the previous iteration has already completed.
if (shouldStop && (await shouldStop())) {
accountLog.info(
{ channel: channelTitle, processed: setIdx, total: archiveSets.length },
"Stop signal received (topic disabled) — skipping remaining archive sets in this scan"
);
break;
}
try { try {
const packageId = await processOneArchiveSet( const packageId = await processOneArchiveSet(
ctx, ctx,