Experiment to rebuild Diffuse using web applets.
feat: more workers
author
Steven Vandevelde
date
(Jul 12, 2025, 10:49 AM +0200)
commit
2cef6867
2cef6867c7557c9465e66adf9a0dec6e645e2fae
parent
47a7750c
47a7750caeedc8cb209ad65d6e85e4743ac67f83
+100
-48
+9
-7
src/pages/engine/queue/_applet.astro
+9
-7
src/pages/engine/queue/_applet.astro
···
1
1
<script>
2
-
import type { Actions } from "@scripts/engine/queue/worker";
2
+
import type { Tasks } from "@scripts/engine/queue/worker";
3
3
import type { Track } from "@applets/core/types";
4
4
import type { State } from "./types.d.ts";
5
5
···
10
10
////////////////////////////////////////////
11
11
// SETUP
12
12
////////////////////////////////////////////
13
-
const worker = endpoint<Actions>(
13
+
const worker = endpoint<Tasks>(
14
14
new SharedWorker(new URL("../../../scripts/engine/queue/worker", import.meta.url), {
15
15
type: "module",
16
16
name: manifest.name,
···
18
18
);
19
19
20
20
// Register applet
21
-
const context = register<State>();
21
+
const context = register<State>({ worker });
22
22
23
23
// Initial state
24
24
context.data = {
···
36
36
context.setActionHandler("unshift", unshift);
37
37
38
38
async function add(items: Track[]) {
39
-
context.data = await worker.add(transfer(context.data), transfer(items));
39
+
context.data = await worker.add(transfer(items));
40
+
console.log(context.data);
40
41
}
41
42
42
43
async function pool(items: Track[]) {
43
-
context.data = await worker.pool(transfer(context.data), transfer(items));
44
+
context.data = await worker.pool(transfer(items));
44
45
}
45
46
46
47
async function shift() {
47
-
context.data = await worker.shift(transfer(context.data));
48
+
context.data = await worker.shift();
49
+
console.log(context.data);
48
50
}
49
51
50
52
async function unshift() {
51
-
context.data = await worker.unshift(transfer(context.data));
53
+
context.data = await worker.unshift();
52
54
}
53
55
</script>
+1
src/pages/orchestrator/input-cache/_applet.astro
+1
src/pages/orchestrator/input-cache/_applet.astro
+3
-1
src/pages/orchestrator/queue-tracks/_applet.astro
+3
-1
src/pages/orchestrator/queue-tracks/_applet.astro
+12
-7
src/pages/processor/metadata/_applet.astro
+12
-7
src/pages/processor/metadata/_applet.astro
···
1
1
<script>
2
-
import type { Actions } from "@scripts/processor/metadata/worker";
2
+
import type { Tasks } from "@scripts/processor/metadata/worker";
3
+
import type { Extraction, Urls } from "./types";
3
4
import { register } from "@scripts/applet/common";
4
5
import { endpoint, SharedWorker, transfer } from "@scripts/common";
5
6
import manifest from "./_manifest.json";
···
7
8
////////////////////////////////////////////
8
9
// SETUP
9
10
////////////////////////////////////////////
10
-
const worker = endpoint<Actions>(
11
+
const worker = endpoint<Tasks>(
11
12
new SharedWorker(new URL("../../../scripts/processor/metadata/worker", import.meta.url), {
12
13
type: "module",
13
14
name: manifest.name,
···
15
16
);
16
17
17
18
// Register applet
18
-
const context = register();
19
+
const context = register({ worker });
19
20
20
21
////////////////////////////////////////////
21
22
// ACTIONS
22
23
////////////////////////////////////////////
23
-
const supply: Actions["supply"] = (args: Parameters<Actions["supply"]>[0]) => {
24
-
const transferredArgs = transfer(args);
25
-
return worker.supply(transferredArgs);
26
-
};
24
+
async function supply(args: {
25
+
includeArtwork?: boolean;
26
+
mimeType?: string;
27
+
stream?: ReadableStream;
28
+
urls?: Urls;
29
+
}): Promise<Extraction> {
30
+
return worker.supply(transfer(args));
31
+
}
27
32
28
33
context.setActionHandler("supply", supply);
29
34
</script>
+3
-3
src/pages/processor/search/_applet.astro
+3
-3
src/pages/processor/search/_applet.astro
···
1
1
<script>
2
-
import type { Actions } from "@scripts/processor/search/worker";
2
+
import type { Tasks } from "@scripts/processor/search/worker";
3
3
import type { Track } from "@applets/core/types";
4
4
import { register } from "@scripts/applet/common";
5
5
import { endpoint, transfer } from "@scripts/common";
···
8
8
////////////////////////////////////////////
9
9
// SETUP
10
10
////////////////////////////////////////////
11
-
const worker = endpoint<Actions>(
11
+
const worker = endpoint<Tasks>(
12
12
new SharedWorker(new URL("../../../scripts/processor/search/worker", import.meta.url), {
13
13
type: "module",
14
14
name: manifest.name,
···
16
16
);
17
17
18
18
// Register applet
19
-
const context = register();
19
+
const context = register({ worker });
20
20
21
21
////////////////////////////////////////////
22
22
// ACTIONS
+51
-22
src/scripts/engine/queue/worker.ts
+51
-22
src/scripts/engine/queue/worker.ts
···
1
1
import type { Track } from "@applets/core/types.js";
2
2
import type { Item, State } from "./types";
3
-
import { arrayShuffle, expose, transfer } from "@scripts/common.ts";
3
+
import { arrayShuffle, provide, transfer } from "@scripts/common.ts";
4
4
5
5
////////////////////////////////////////////
6
6
// STATE
···
12
12
pool: [],
13
13
};
14
14
15
+
const state: State = {
16
+
future: [],
17
+
past: [],
18
+
now: null,
19
+
};
20
+
21
+
function data() {
22
+
return transfer({ ...state });
23
+
}
24
+
15
25
////////////////////////////////////////////
16
-
// ACTIONS
26
+
// SETUP
17
27
////////////////////////////////////////////
18
-
const actions = expose({
28
+
29
+
const actions = {
19
30
add,
20
31
pool,
21
32
shift,
22
33
unshift,
34
+
};
35
+
36
+
const { tasks } = provide({
37
+
actions,
38
+
tasks: actions,
23
39
});
24
40
25
41
export type Actions = typeof actions;
42
+
export type Tasks = typeof tasks;
26
43
27
-
// Actions
44
+
////////////////////////////////////////////
45
+
// ACTIONS
46
+
////////////////////////////////////////////
28
47
29
-
function add(state: State, items: Item[]): State {
30
-
return transfer({ ...state, future: [...state.future, ...items] });
48
+
function add(items: Item[]): State {
49
+
state.future = [...state.future, ...items];
50
+
51
+
// Fin
52
+
return data();
31
53
}
32
54
33
-
function pool(state: State, tracks: Track[]): State {
55
+
function pool(tracks: Track[]): State {
34
56
internal.pool = tracks;
35
57
36
58
// TODO: If the pool changes, only remove non-existing tracks
···
38
60
//
39
61
// What about past queue items?
40
62
41
-
state = fill({ ...state, future: [] });
63
+
state.future = [];
64
+
fill();
42
65
43
66
// Automatically insert track if there isn't any
44
-
if (!state.now) return shift(state);
45
-
return transfer(state);
67
+
if (!state.now) return shift();
68
+
69
+
// Fin
70
+
return data();
46
71
}
47
72
48
-
function shift(state: State): State {
49
-
const now = state.future[0] || null;
50
-
const future = state.future.slice(1);
51
-
const past = state.now ? [...state.past, state.now] : state.past;
52
-
const filled = fill({ past, now, future });
73
+
function shift(): State {
74
+
state.now = state.future[0] || null;
75
+
state.future = state.future.slice(1);
76
+
state.past = state.now ? [...state.past, state.now] : state.past;
53
77
54
-
return transfer(filled);
78
+
fill();
79
+
80
+
// Fin
81
+
return data();
55
82
}
56
83
57
-
function unshift(state: State): State {
84
+
function unshift(): State {
58
85
if (state.past.length === 0) return state;
59
86
60
87
const past = [...state.past];
61
88
const [last] = past.splice(past.length - 1, 1);
62
-
const now = last ?? null;
63
-
const future = state.now ? [state.now, ...state.future] : state.future;
89
+
state.now = last ?? null;
90
+
state.future = state.now ? [state.now, ...state.future] : state.future;
64
91
65
-
return transfer({ past, now, future });
92
+
// Fin
93
+
return data();
66
94
}
67
95
68
96
// 🛠️
69
97
70
98
// TODO: Most likely there's a more performant solution
71
-
function fill(state: State): State {
99
+
function fill() {
72
100
if (state.future.length >= QUEUE_SIZE) return state;
73
101
74
102
let reducedPool = internal.pool.reduce(
···
92
120
}
93
121
94
122
const poolSelection = arrayShuffle(reducedPool).slice(0, QUEUE_SIZE - state.future.length);
95
-
return add(state, poolSelection);
123
+
124
+
add(poolSelection);
96
125
}
+8
-2
src/scripts/processor/metadata/worker.ts
+8
-2
src/scripts/processor/metadata/worker.ts
···
1
1
import type { Extraction, Urls } from "./types.d.ts";
2
-
import { expose, transfer } from "@scripts/common";
2
+
import { provide, transfer } from "@scripts/common";
3
3
import { musicMetadataTags } from "./common.ts";
4
4
5
5
////////////////////////////////////////////
6
6
// ACTIONS
7
7
////////////////////////////////////////////
8
-
const actions = expose({
8
+
const actions = {
9
9
supply,
10
+
};
11
+
12
+
const { tasks } = provide({
13
+
actions,
14
+
tasks: actions,
10
15
});
11
16
12
17
export type Actions = typeof actions;
18
+
export type Tasks = typeof tasks;
13
19
14
20
// Actions
15
21
+13
-6
src/scripts/processor/search/worker.ts
+13
-6
src/scripts/processor/search/worker.ts
···
2
2
// import { pluginQPS } from "@orama/plugin-qps";
3
3
4
4
import type { Track } from "@applets/core/types";
5
-
import { expose, transfer } from "@scripts/common";
5
+
import { expose, provide, transfer } from "@scripts/common";
6
6
import { SCHEMA } from "./constants";
7
7
import type { State } from "./types";
8
8
···
52
52
// },
53
53
});
54
54
55
-
////////////////////////////////////////////
56
-
// ACTIONS
57
-
////////////////////////////////////////////
58
-
const actions = expose({
55
+
// 🚀
56
+
57
+
const actions = {
59
58
search,
60
59
supply,
60
+
};
61
+
62
+
const { tasks } = provide({
63
+
actions,
64
+
tasks: actions,
61
65
});
62
66
63
67
export type Actions = typeof actions;
68
+
export type Tasks = typeof tasks;
64
69
65
-
// Actions
70
+
////////////////////////////////////////////
71
+
// ACTIONS
72
+
////////////////////////////////////////////
66
73
67
74
async function search(term: string): Promise<Track[]> {
68
75
const results = await Orama.search(db, {