File size: 4,544 Bytes
4e23b01 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | import { parentPort, workerData } from 'node:worker_threads';
import { configureTextBuildWorkerRuntime } from '@moonshot-ai/minidb/worker-runtime';
import { GlobalSearchError, type GlobalSearchErrorReason } from '../contract.ts';
import {
SearchIndexCore,
type CoreSearchParams,
type SyncSessionInput,
} from '../indexCore.ts';
import {
SEARCH_WORKER_PROTOCOL_VERSION,
type SearchWorkerData,
type SearchWorkerErrorPayload,
type SearchWorkerEvent,
type SearchWorkerOpenResult,
type SearchWorkerCall,
type SearchWorkerRequest,
} from './protocol.ts';
const data = workerData as SearchWorkerData;
if (typeof data.textBuildWorkerPath === 'string') {
try {
configureTextBuildWorkerRuntime(data.textBuildWorkerPath);
} catch {
}
}
const port = parentPort;
if (port === null) {
throw new Error('search worker entry must run inside a worker thread');
}
const post = (event: SearchWorkerEvent): void => {
port.postMessage(event);
};
const core = new SearchIndexCore({
indexDir: data.dir,
bootSalt: data.bootSalt,
log: {
info: (message, meta) => {
post({ type: 'log', level: 'info', message, meta });
},
warn: (message, meta) => {
post({ type: 'log', level: 'warn', message, meta });
},
},
onLockToken: (token) => {
post({ type: 'lockToken', token });
},
});
function toErrorPayload(error: unknown): SearchWorkerErrorPayload {
if (error instanceof GlobalSearchError) {
return { message: error.message, reason: error.reason as GlobalSearchErrorReason };
}
return { message: error instanceof Error ? error.message : String(error) };
}
async function dispatch(request: SearchWorkerCall): Promise<unknown> {
switch (request.type) {
case 'open': {
await core.ensureOpen();
const result: SearchWorkerOpenResult = {
readOnly: core.db?.readOnly === true,
lockToken: core.lockTokenView,
lifecycle: core.lifecycleState(),
};
return result;
}
case 'search':
return core.search(request.params as CoreSearchParams);
case 'sync':
return core.sync((request.params as { sessions: readonly SyncSessionInput[] }).sessions);
case 'refresh': {
await core.refresh();
const result: SearchWorkerOpenResult = {
readOnly: core.db?.readOnly === true,
lockToken: core.lockTokenView,
lifecycle: core.lifecycleState(),
};
return result;
}
case 'reindex': {
await core.reindex();
const result: SearchWorkerOpenResult = {
readOnly: core.db?.readOnly === true,
lockToken: core.lockTokenView,
lifecycle: core.lifecycleState(),
};
return result;
}
case 'status':
return core.status();
case 'close':
return null;
}
}
const inFlight = new Set<Promise<void>>();
let closing = false;
async function handle(request: SearchWorkerCall): Promise<void> {
if (request.v !== SEARCH_WORKER_PROTOCOL_VERSION) {
post({
id: request.id,
type: 'error',
error: {
message: `search worker protocol mismatch: host v${request.v}, worker v${SEARCH_WORKER_PROTOCOL_VERSION}`,
},
});
return;
}
if (request.type === 'close') {
closing = true;
core.beginClose();
await Promise.all(inFlight);
let error: SearchWorkerErrorPayload | null = null;
try {
await core.close();
} catch (closeError) {
error = toErrorPayload(closeError);
}
if (error !== null) post({ id: request.id, type: 'error', error });
else post({ id: request.id, type: 'result', result: null });
port!.close();
return;
}
if (closing) {
post({
id: request.id,
type: 'error',
error: { message: 'search service is disposed', reason: 'index_unavailable' },
});
return;
}
try {
const result = await dispatch(request);
post({ id: request.id, type: 'result', result });
} catch (error) {
post({ id: request.id, type: 'error', error: toErrorPayload(error) });
}
}
port.on('message', (value: unknown) => {
const request = value as SearchWorkerRequest;
if (request === null || typeof request !== 'object') return;
if (request.type === 'beginClose') {
closing = true;
core.beginClose();
return;
}
if (typeof request.id !== 'number' || typeof request.type !== 'string') return;
const tracked = handle(request);
inFlight.add(tracked);
void tracked.finally(() => {
inFlight.delete(tracked);
});
});
post({ type: 'ready', v: SEARCH_WORKER_PROTOCOL_VERSION });
|