diff --git a/src/plugins/built-in/globalSearch/src/indexing/indexer.ts b/src/plugins/built-in/globalSearch/src/indexing/indexer.ts index 7d12084c..8aa368cb 100644 --- a/src/plugins/built-in/globalSearch/src/indexing/indexer.ts +++ b/src/plugins/built-in/globalSearch/src/indexing/indexer.ts @@ -1,4 +1,4 @@ -import { clear, getAll, put, remove } from "./db"; +import { clear, getAll, get, put, remove } from "./db"; import { jobs } from "./jobs"; import { renderComponentMap } from "./renderComponents"; import type { HydratedIndexItem, IndexItem, Job, JobContext } from "./types"; @@ -9,6 +9,17 @@ const LOCK_KEY = "bsq-indexer-lock"; const HEARTBEAT_INTERVAL = 10000; const LOCK_TIMEOUT = 20000; +/* ─────────── Progress‑meta helpers ─────────── */ +async function loadProgress(jobId: string): Promise { + const rec = await get(META_STORE, `progress:${jobId}`); + return rec?.progress as T | undefined; +} + +async function saveProgress(jobId: string, progress: T): Promise { + await put(META_STORE, { jobId, progress }, `progress:${jobId}`); +} +/* ───────────────────────────────────────────── */ + let heartbeatTimer: ReturnType | null = null; function shouldRun(job: Job, lastRun?: number): boolean { @@ -132,26 +143,27 @@ export async function runIndexing(): Promise { continue; } - // These DB operations happen on the main thread (acceptable per request) - const getStoredItems = async () => await getAll(jobId); - const setStoredItems = async (items: IndexItem[]) => { - await clear(jobId); - // Add validation before putting + const getStoredItems = async (storeId?: string) => await getAll(storeId ?? jobId); + const setStoredItems = async (items: IndexItem[], storeId?: string) => { + const targetStore = storeId ?? jobId; + await clear(targetStore); const validItems = items.filter(i => i && i.id); if (validItems.length !== items.length) { - console.warn(`[Indexer Job ${jobId}] Filtered out ${items.length - validItems.length} invalid items before storing.`); + console.warn(`[Indexer Job ${jobId} -> Store ${targetStore}] Filtered out ${items.length - validItems.length} invalid items before storing.`); } - await Promise.all(validItems.map((i) => put(jobId, i, i.id))); + await Promise.all(validItems.map((i) => put(targetStore, i, i.id))); }; - const addItem = async (item: IndexItem) => { - if (item && item.id) { // Add validation - await put(jobId, item, item.id); + const addItem = async (item: IndexItem, storeId?: string) => { + const targetStore = storeId ?? jobId; + if (item && item.id) { + await put(targetStore, item, item.id); } else { - console.warn(`[Indexer Job ${jobId}] Attempted to add invalid item:`, item); + console.warn(`[Indexer Job ${jobId} -> Store ${targetStore}] Attempted to add invalid item:`, item); } }; - const removeItem = async (id: string) => { - await remove(jobId, id); + const removeItem = async (id: string, storeId?: string) => { + const targetStore = storeId ?? jobId; + await remove(targetStore, id); }; const ctx: JobContext = { @@ -159,6 +171,8 @@ export async function runIndexing(): Promise { setStoredItems, addItem, removeItem, + getProgress: () => loadProgress(jobId), + setProgress: (p) => saveProgress(jobId, p), }; console.debug(`%c[Indexer] Running job "${jobId}"...`, "color: #4ea1ff"); @@ -195,11 +209,11 @@ export async function runIndexing(): Promise { allItemsFromJobs.push(...hydratedItems); console.debug( - `%c[Indexer] ${job.label}: ${newItemsRaw.length} new items fetched, ${merged.length} total stored (non-vector).`, + `%c[Indexer] ${job.label}: ${newItemsRaw.length} new items from run, ${merged.length} total stored in '${jobId}' store (non-vector).`, "color: #00c46f", ); } catch (err) { - console.debug(`%c[Indexer] ${job.label} failed:`, "color: red"); + console.debug(`%c[Indexer] Job ${job.label} failed:`, "color: red"); console.error(err); } diff --git a/src/plugins/built-in/globalSearch/src/indexing/jobs.ts b/src/plugins/built-in/globalSearch/src/indexing/jobs.ts index 1ecc1675..b5b76d17 100644 --- a/src/plugins/built-in/globalSearch/src/indexing/jobs.ts +++ b/src/plugins/built-in/globalSearch/src/indexing/jobs.ts @@ -1,350 +1,8 @@ import type { Job } from "./types"; -import type { IndexItem } from "./types"; - -interface MessageNotification { - notificationID: number; - type: "message"; - message: { - subtitle: string; - messageID: number; - title: string; - }; - timestamp: string; -} - -interface AssessmentNotification { - notificationID: number; - type: "coneqtassessments"; - coneqtAssessments: { - programmeID: number; - metaclassID: number; - subtitle: string; - term: string; - title: string; - assessmentID: number; - subjectCode: string; - }; - timestamp: string; -} - -type Notification = MessageNotification | AssessmentNotification; - -interface MessageListResponse { - payload: { - hasMore: boolean; - messages: { - date: string; - attachments: boolean; - attachmentCount: number; - read: number; - sender: string; - sender_id: number; - sender_type: string; - subject: string; - id: number; - participants: Array<{ - name: string; - photo: string; - type: string; - }>; - }[]; - ts: string; - }; - status: string; -} - -interface MessageContentResponse { - payload: { - date: string; - blind: boolean; - read: boolean; - subject: string; - sender_type: string; - sender_id: number; - starred: boolean; - contents: string; - sender: string; - files: any[]; - id: number; - participants: Array<{ - read: number; - name: string; - photo: string; - id: number; - type: string; - }>; - }; - status: string; -} - -// Helper to strip HTML tags from text -function stripHtmlTags(html: string): string { - return html.replace(/<[^>]*>/g, ""); -} - -// Helper to fetch messages with pagination -async function fetchMessages( - offset: number = 0, - limit: number = 100, -): Promise { - const response = await fetch( - `${location.origin}/seqta/student/load/message`, - { - method: "POST", - credentials: "include", - headers: { - "Content-Type": "application/json; charset=utf-8", - }, - body: JSON.stringify({ - searchValue: "", - sortBy: "date", - sortOrder: "desc", - action: "list", - label: "inbox", - offset, - limit, - datetimeUntil: null, - }), - }, - ); - - return await response.json(); -} - -// Helper to fetch message content -async function fetchMessageContent( - messageId: number, -): Promise { - const response = await fetch( - `${location.origin}/seqta/student/load/message`, - { - method: "POST", - credentials: "include", - headers: { - "Content-Type": "application/json; charset=utf-8", - }, - body: JSON.stringify({ - action: "message", - id: messageId, - }), - }, - ); - - return await response.json(); -} - -// Helper to fetch notifications -async function fetchNotifications(): Promise { - const response = await fetch(`${location.origin}/seqta/student/heartbeat?`, { - method: "POST", - headers: { - "Content-Type": "application/json; charset=utf-8", - }, - body: JSON.stringify({ - timestamp: "1970-01-01 00:00:00.0", - hash: "#?page=/notifications", - }), - }); - - const json = await response.json(); - return json.notifications ?? []; -} +import { messagesJob } from "./jobs/messages"; +import { assessmentsJob } from "./jobs/assessments"; export const jobs: Record = { - messages: { - id: "messages", - label: "Messages", - renderComponentId: "message", - frequency: { type: "expiry", afterMs: 1000 }, // every 5 minutes - - run: async (ctx) => { - // Get existing items first - const existing = await ctx.getStoredItems(); - const existingIds = new Set(existing.map((i) => i.id)); - const newItems: IndexItem[] = []; - let offset = 0; - const limit = 100; - let hasMore = true; - let consecutiveExisting = 0; - - // Fetch all messages with pagination - while (hasMore) { - try { - const response = await fetchMessages(offset, limit); - - if (response.status !== "200") { - console.error("Failed to fetch messages:", response); - break; - } - - const messages = response.payload.messages; - hasMore = response.payload.hasMore; - - // Process each message - for (const message of messages) { - const id = message.id.toString(); - - // Skip if we already have this message - if (existingIds.has(id)) { - consecutiveExisting++; - // If we've found 20 consecutive existing messages, assume we've caught up - if (consecutiveExisting >= 20) { - console.debug( - "[Messages Job] Found 20 consecutive existing messages, stopping fetch", - ); - hasMore = false; - break; - } - continue; - } - - // Reset consecutive counter when we find a new message - consecutiveExisting = 0; - - try { - // Fetch message content - const contentResponse = await fetchMessageContent(message.id); - - if (contentResponse.status !== "200") { - console.error( - "Failed to fetch message content:", - contentResponse, - ); - continue; - } - - const content = stripHtmlTags(contentResponse.payload.contents); - - newItems.push({ - id, - text: message.subject, - category: "messages", - content: `From: ${message.sender}\n\n${content}`, - dateAdded: new Date(message.date).getTime(), - metadata: { - messageId: message.id, - author: message.sender, - senderId: message.sender_id, - senderType: message.sender_type, - timestamp: message.date, - hasAttachments: message.attachments, - attachmentCount: message.attachmentCount, - read: message.read === 1, - }, - actionId: "message", - renderComponentId: "message", - }); - - // Add to existingIds as we process to prevent duplicates in the same run - existingIds.add(id); - } catch (error) { - console.error("Error fetching message content:", error); - continue; - } - } - - offset += limit; - } catch (error) { - console.error("Error fetching messages:", error); - break; - } - - // Small delay to avoid overwhelming the server - await new Promise((resolve) => setTimeout(resolve, 100)); - } - - console.debug(`[Messages Job] Found ${newItems.length} new messages`); - return newItems; - }, - - purge: (items) => { - const cutoff = Date.now() - 4 * 12 * 30 * 24 * 60 * 60 * 1000; - return items.filter((i) => i.dateAdded >= cutoff); - }, - }, - - assessments: { - id: "assessments", - label: "Assessments", - renderComponentId: "assessment", - frequency: { type: "expiry", afterMs: 1000 * 60 * 15 }, // every 15 minutes - - run: async (ctx) => { - const notifications = await fetchNotifications(); - const assessmentNotifications = notifications.filter( - (n): n is MessageNotification | AssessmentNotification => - n.type === "coneqtassessments" || - (n.type === "message" && - n.message.title.toLowerCase().includes("assessment")), - ); - - const existing = await ctx.getStoredItems(); - const existingIds = new Set(existing.map((i) => i.id)); - const newItems: IndexItem[] = []; - - for (const notification of assessmentNotifications) { - const id = notification.notificationID.toString(); - if (existingIds.has(id)) continue; - - if (notification.type === "coneqtassessments") { - const { coneqtAssessments: assessment } = notification; - newItems.push({ - id, - text: assessment.title, - category: "assessments", - content: assessment.subtitle, - dateAdded: new Date(notification.timestamp).getTime(), - metadata: { - assessmentId: assessment.assessmentID, - subject: assessment.subjectCode, - term: assessment.term, - programmeId: assessment.programmeID, - metaclassId: assessment.metaclassID, - timestamp: notification.timestamp, - }, - actionId: "assessment", - renderComponentId: "assessment", - }); - } else { - // Handle message-based assessments - const { message } = notification; - newItems.push({ - id, - text: message.title, - category: "assessments", - content: `From: ${message.subtitle}`, - dateAdded: new Date(notification.timestamp).getTime(), - metadata: { - messageId: message.messageID, - author: message.subtitle, - timestamp: notification.timestamp, - isMessageBased: true, - }, - actionId: "assessment", - renderComponentId: "assessment", - }); - } - } - - return newItems; - }, - - purge: (items) => { - // Keep assessments from the current year - const date = new Date(); - date.setMonth(0); // January - date.setDate(1); - date.setHours(0); - date.setMinutes(0); - date.setSeconds(0); - const cutoff = date.getTime(); - return items.filter((i) => i.dateAdded >= cutoff); - }, - }, - - // We can add more job types here as needed: - // - notices - // - timetable changes - // - homework - // etc. -}; + messages: messagesJob, + assessments: assessmentsJob, +}; \ No newline at end of file diff --git a/src/plugins/built-in/globalSearch/src/indexing/jobs/assessments.ts b/src/plugins/built-in/globalSearch/src/indexing/jobs/assessments.ts new file mode 100644 index 00000000..d88f4566 --- /dev/null +++ b/src/plugins/built-in/globalSearch/src/indexing/jobs/assessments.ts @@ -0,0 +1,138 @@ +import type { Job, IndexItem } from "../types"; + +/* ------------- Notification types ------------- */ +interface MessageNotification { + notificationID: number; + type: "message"; + message: { subtitle: string; messageID: number; title: string }; + timestamp: string; +} + +interface AssessmentNotification { + notificationID: number; + type: "coneqtassessments"; + coneqtAssessments: { + programmeID: number; + metaclassID: number; + subtitle: string; + term: string; + title: string; + assessmentID: number; + subjectCode: string; + }; + timestamp: string; +} + +type Notification = MessageNotification | AssessmentNotification; + +/* ------------- Progress model ------------- */ +interface AssessmentsProgress { + lastTs: number; // ms since epoch of last processed notification +} + +/* ------------- Helpers ------------- */ +const fetchNotifications = async () => { + const res = await fetch(`${location.origin}/seqta/student/heartbeat?`, { + method: "POST", + headers: { "Content-Type": "application/json; charset=utf-8" }, + body: JSON.stringify({ + timestamp: "1970-01-01 00:00:00.0", + hash: "#?page=/notifications", + }), + }); + const json = await res.json(); + return (json.notifications ?? []) as Notification[]; +}; + +/* ------------- Job ------------- */ +export const assessmentsJob: Job = { + id: "assessments", + label: "Assessments", + renderComponentId: "assessment", + frequency: { type: "expiry", afterMs: 15 * 60 * 1000 }, + + run: async (ctx) => { + const progress = + (await ctx.getProgress()) ?? { lastTs: 0 }; + + let notifications: Notification[]; + try { + notifications = await fetchNotifications(); + } catch (e) { + console.error("[Assessments job] fetch failed:", e); + return []; + } + + const notificationIsIndexed = async (id: string): Promise => { + const [inAssessments, inMessages] = await Promise.all([ + ctx.getStoredItems("assessments").then((items) => items.some((i) => i.id === id)), + ctx.getStoredItems("messages").then((items) => items.some((i) => i.id === id)), + ]); + return inAssessments || inMessages; + }; + + const items: IndexItem[] = []; + + for (const notif of notifications) { + const id = notif.notificationID.toString(); + if (await notificationIsIndexed(id)) continue; + + if (notif.type === "coneqtassessments") { + const a = notif.coneqtAssessments; + items.push({ + id, + text: a.title, + category: "assessments", + content: a.subtitle, + dateAdded: new Date(notif.timestamp).getTime(), + metadata: { + assessmentId: a.assessmentID, + subject: a.subjectCode, + term: a.term, + programmeId: a.programmeID, + metaclassId: a.metaclassID, + timestamp: notif.timestamp, + }, + actionId: "assessment", + renderComponentId: "assessment", + }); + } else { + await ctx.addItem( + { + id, + text: notif.message.title, + category: "messages", + content: `From: ${notif.message.subtitle}`, + dateAdded: new Date(notif.timestamp).getTime(), + metadata: { + messageId: notif.message.messageID, + author: notif.message.subtitle, + timestamp: notif.timestamp, + isAssessmentNotification: true, + }, + actionId: "message", + renderComponentId: "message", + }, + "messages" + ); + } + } + + if (items.length) { + const latest = Math.max( + ...items.map((i) => i.dateAdded), + progress.lastTs, + ); + await ctx.setProgress({ lastTs: latest }); + } + + return items; + }, + + purge: (items) => { + const date = new Date(); + date.setMonth(0, 1); + date.setHours(0, 0, 0, 0); + return items.filter((i) => i.dateAdded >= date.getTime()); + }, +}; \ No newline at end of file diff --git a/src/plugins/built-in/globalSearch/src/indexing/jobs/messages.ts b/src/plugins/built-in/globalSearch/src/indexing/jobs/messages.ts new file mode 100644 index 00000000..eb32358c --- /dev/null +++ b/src/plugins/built-in/globalSearch/src/indexing/jobs/messages.ts @@ -0,0 +1,133 @@ +import type { Job, IndexItem } from "../types"; + +const stripHtmlTags = (html: string) => html.replace(/<[^>]*>/g, ""); + +const fetchMessages = async (offset = 0, limit = 100) => { + const res = await fetch(`${location.origin}/seqta/student/load/message`, { + method: "POST", + credentials: "include", + headers: { "Content-Type": "application/json; charset=utf-8" }, + body: JSON.stringify({ + searchValue: "", + sortBy: "date", + sortOrder: "desc", + action: "list", + label: "inbox", + offset, + limit, + datetimeUntil: null, + }), + }); + return res.json() as Promise<{ + payload: { hasMore: boolean; messages: any[]; ts: string }; + status: string; + }>; +}; + +const fetchMessageContent = async (id: number) => { + const res = await fetch(`${location.origin}/seqta/student/load/message`, { + method: "POST", + credentials: "include", + headers: { "Content-Type": "application/json; charset=utf-8" }, + body: JSON.stringify({ action: "message", id }), + }); + return res.json() as Promise<{ + payload: { contents: string }; + status: string; + }>; +}; + +interface MessagesProgress { + offset: number; + done: boolean; +} + +export const messagesJob: Job = { + id: "messages", + label: "Messages", + renderComponentId: "message", + frequency: { type: "expiry", afterMs: 1000 * 60 * 60 * 24 }, + + run: async (ctx) => { + const limit = 100; + const progress = + (await ctx.getProgress()) ?? { offset: 0, done: false }; + + const existingIds = new Set( + (await ctx.getStoredItems()).map((i) => i.id), + ); + + let consecutiveExisting = 0; + + while (!progress.done) { + let list; + try { + list = await fetchMessages(progress.offset, limit); + } catch (e) { + console.error("[Messages job] list fetch failed:", e); + break; + } + + if (list.status !== "200") break; + + for (const msg of list.payload.messages) { + const id = msg.id.toString(); + + if (existingIds.has(id)) { + consecutiveExisting += 1; + if (consecutiveExisting >= 20) { + progress.done = true; + break; + } + continue; + } + consecutiveExisting = 0; + + let full; + try { + full = await fetchMessageContent(msg.id); + } catch (e) { + console.error(`[Messages job] content fetch failed (id ${id}):`, e); + continue; + } + if (full.status !== "200") continue; + + const item: IndexItem = { + id, + text: msg.subject, + category: "messages", + content: `From: ${msg.sender}\n\n${stripHtmlTags(full.payload.contents)}`, + dateAdded: new Date(msg.date).getTime(), + metadata: { + messageId: msg.id, + author: msg.sender, + senderId: msg.sender_id, + senderType: msg.sender_type, + timestamp: msg.date, + hasAttachments: msg.attachments, + attachmentCount: msg.attachmentCount, + read: msg.read === 1, + }, + actionId: "message", + renderComponentId: "message", + }; + + await ctx.addItem(item); + existingIds.add(id); + } + + if (!list.payload.hasMore) progress.done = true; + progress.offset += limit; + await ctx.setProgress(progress); + } + + if (progress.done) await ctx.setProgress({ offset: 0, done: false }); + + return []; + }, + + purge: (items) => { + const fourYears = Date.now() - 4 * 365 * 24 * 60 * 60 * 1000; + return items.filter((i) => i.dateAdded >= fourYears); + }, +}; \ No newline at end of file diff --git a/src/plugins/built-in/globalSearch/src/indexing/types.ts b/src/plugins/built-in/globalSearch/src/indexing/types.ts index 094757b4..098e1238 100644 --- a/src/plugins/built-in/globalSearch/src/indexing/types.ts +++ b/src/plugins/built-in/globalSearch/src/indexing/types.ts @@ -21,10 +21,12 @@ export type Frequency = | { type: "expiry"; afterMs: number }; export interface JobContext { - getStoredItems: () => Promise; - setStoredItems: (items: IndexItem[]) => Promise; - addItem: (item: IndexItem) => Promise; - removeItem: (id: string) => Promise; + getStoredItems: (storeId?: string) => Promise; + setStoredItems: (items: IndexItem[], storeId?: string) => Promise; + addItem: (item: IndexItem, storeId?: string) => Promise; + removeItem: (id: string, storeId?: string) => Promise; + getProgress: () => Promise; + setProgress: (progress: T) => Promise; } export interface Job { @@ -34,4 +36,4 @@ export interface Job { renderComponentId: string; run: (ctx: JobContext) => Promise; purge?: (items: IndexItem[]) => IndexItem[]; -} +} \ No newline at end of file