Skip to content

Commit ecb306d

Browse files
kallanreedkallanreed
committed
Move all samples to use D1
1 parent 7d6d7d0 commit ecb306d

11 files changed

Lines changed: 252 additions & 164 deletions

File tree

‎content/shared.js‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1098,8 +1098,8 @@ function parseLocation(latStr, lonStr) {
10981098
}
10991099
return [lat, lon];
11001100
}
1101+
var dayInMillis = 24 * 60 * 60 * 1e3;
11011102
function ageInDays(time) {
1102-
const dayInMillis = 24 * 60 * 60 * 1e3;
11031103
return (Date.now() - new Date(time)) / dayInMillis;
11041104
}
11051105
function pushMap(map, key, value) {
@@ -1169,6 +1169,7 @@ export {
11691169
and,
11701170
centerPos,
11711171
coverageKey,
1172+
dayInMillis,
11721173
definedOr,
11731174
fadeColor,
11741175
fromTruncatedTime,

‎content/shared_npm.js‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -74,8 +74,9 @@ export function parseLocation(latStr, lonStr) {
7474
return [lat, lon];
7575
}
7676

77+
export const dayInMillis = 24 * 60 * 60 * 1000;
78+
7779
export function ageInDays(time) {
78-
const dayInMillis = 24 * 60 * 60 * 1000;
7980
return (Date.now() - new Date(time)) / dayInMillis;
8081
}
8182

‎functions/clean-up.js‎

Lines changed: 1 addition & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -28,17 +28,6 @@ async function cleanCoverage(context, result) {
2828
}
2929

3030
async function cleanSamples(context, result) {
31-
// // This should mostly be done by consolidate.js
32-
// const store = context.env.SAMPLES;
33-
// let cursor = null;
34-
35-
// do {
36-
// const samples = await store.list({ cursor: cursor });
37-
// cursor = samples.cursor ?? null;
38-
39-
// for (const key of samples.keys) {
40-
// }
41-
// } while (cursor !== null);
4231
}
4332

4433
function overlaps(a, b) {
@@ -154,6 +143,7 @@ export async function onRequest(context) {
154143

155144
case "repeaters":
156145
await cleanRepeaters(context, result);
146+
break;
157147
}
158148

159149
return Response.json(result);

‎functions/consolidate.js‎

Lines changed: 52 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,10 @@
22
import * as util from '../content/shared.js';
33

44
// TODO: App-token for 'auth'?
5+
// TODO: More of this could be handled in SQL.
56

6-
// Samples are consolidated after they are this age.
7-
const DEF_CONSOLIDATE_AGE = 0.5;
7+
// Samples are consolidated after they are this age in days.
8+
const DEF_CONSOLIDATE_AGE = 1;
89

910
// Only the N-newest samples are kept so that
1011
// recent samples can eventually flip a coverage tile.
@@ -30,29 +31,27 @@ function consolidateSamples(samples, cutoffTime) {
3031
// Build the uber sample.
3132
samples.forEach(s => {
3233
// Was this sample handled in a previous batch?
33-
if (s.metadata.time <= cutoffTime)
34+
if (s.time <= cutoffTime)
3435
return;
3536

36-
const path = s.metadata.path ?? [];
37-
const observed = s.metadata.observed ?? path.length > 0;
37+
uberSample.time = Math.max(s.time, uberSample.time);
38+
uberSample.snr = util.definedOr(Math.max, s.snr, uberSample.snr);
39+
uberSample.rssi = util.definedOr(Math.max, s.rssi, uberSample.rssi);
3840

39-
uberSample.time = Math.max(s.metadata.time, uberSample.time);
40-
uberSample.snr = util.definedOr(Math.max, s.metadata.snr, uberSample.snr);
41-
uberSample.rssi = util.definedOr(Math.max, s.metadata.rssi, uberSample.rssi);
42-
43-
if (observed) {
41+
if (s.observed) {
4442
uberSample.observed++;
45-
uberSample.lastObserved = Math.max(s.metadata.time, uberSample.lastObserved);
43+
uberSample.lastObserved = Math.max(s.time, uberSample.lastObserved);
4644
}
4745

48-
if (path.length > 0) {
46+
if (s.observed || s.repeaters.length > 0) {
4947
uberSample.heard++;
50-
uberSample.lastHeard = Math.max(s.metadata.time, uberSample.lastHeard);
48+
uberSample.lastHeard = Math.max(s.time, uberSample.lastHeard);
5149
} else {
5250
uberSample.lost++;
5351
}
5452

55-
path.forEach(p => {
53+
const repeaters = JSON.parse(s.repeaters);
54+
repeaters.forEach(p => {
5655
if (!uberSample.repeaters.includes(p))
5756
uberSample.repeaters.push(p);
5857
});
@@ -108,7 +107,7 @@ async function mergeCoverage(key, samples, store) {
108107
// Sort and keep the N-newest.
109108
value = value.toSorted((a, b) => a.time - b.time).slice(-MAX_SAMPLES_PER_COVERAGE);
110109
}
111-
110+
112111
// Compute new metadata stats, but keep the existing repeater list (for now).
113112
const metadata = {
114113
observed: 0,
@@ -139,52 +138,47 @@ async function mergeCoverage(key, samples, store) {
139138

140139
export async function onRequest(context) {
141140
const coverageStore = context.env.COVERAGE;
142-
const sampleStore = context.env.SAMPLES;
143-
const archiveStore = context.env.ARCHIVE;
144141

145142
const url = new URL(context.request.url);
146143
let maxAge = url.searchParams.get('maxAge') ?? DEF_CONSOLIDATE_AGE; // Days
147144
if (maxAge <= 0)
148145
maxAge = DEF_CONSOLIDATE_AGE;
149146

150147
const result = {
151-
coverage_entites_to_update: 0,
148+
coverage_to_update: 0,
152149
samples_to_update: 0,
153150
merged_ok: 0,
154151
merged_fail: 0,
155-
archive_ok: 0,
156-
archive_fail: 0,
157-
delete_ok: 0,
158-
delete_fail: 0,
159-
delete_skip: 0
152+
merged_skip: 0,
160153
};
154+
const now = Date.now();
161155
const hashToSamples = new Map();
162-
let cursor = null;
163-
164-
// Build index of old samples.
165-
do {
166-
const samplesList = await sampleStore.list({ cursor: cursor });
167-
cursor = samplesList.cursor ?? null;
168-
169-
// Group samples by 6-digit hash
170-
samplesList.keys.forEach(s => {
171-
// Ignore recent samples.
172-
if (util.ageInDays(s.metadata.time) < maxAge) return;
173-
174-
result.samples_to_update++;
175-
const key = s.name.substring(0, 6);
176-
util.pushMap(hashToSamples, key, {
177-
key: s.name,
178-
metadata: s.metadata
179-
});
180-
});
181-
} while (cursor !== null);
182156

183-
result.coverage_entites_to_update = hashToSamples.size
157+
// Get old samples.
158+
const { results: samples } = await context.env.DB
159+
.prepare("SELECT * FROM samples WHERE time < ?")
160+
.bind(now - (maxAge * util.dayInMillis))
161+
.all();
162+
console.log(`Old samples:${samples.length}`);
163+
result.samples_to_update = samples.length;
164+
165+
// Build index of old samples - group by 6-digit hash.
166+
samples.forEach(s => {
167+
const key = s.hash.substring(0, 6);
168+
util.pushMap(hashToSamples, key, s);
169+
});
170+
console.log(`Coverage to update:${hashToSamples.size}`);
171+
result.coverage_to_update = hashToSamples.size
172+
184173
const mergedKeys = [];
174+
let mergeCount = 0;
185175

186176
// Merge old samples into coverage items.
187-
await Promise.all(hashToSamples.entries().map(async ([k, v]) => {
177+
for (const [k, v] of hashToSamples.entries()) {
178+
// To prevent hitting KV limits, only handle first N.
179+
if (++mergeCount > 500)
180+
break;
181+
188182
try {
189183
await mergeCoverage(k, v, coverageStore);
190184
result.merged_ok++;
@@ -193,31 +187,24 @@ export async function onRequest(context) {
193187
console.log(`Merge failed. ${e}`);
194188
result.merged_fail++;
195189
}
196-
}));
190+
}
191+
result.merged_skip = hashToSamples.size - (result.merged_ok + result.merged_fail);
197192

198193
// Archive and delete the old samples.
199-
await Promise.all(mergedKeys.map(async k => {
194+
const cleanupStmts = [];
195+
mergedKeys.forEach(k => {
200196
const v = hashToSamples.get(k);
201197
for (const sample of v) {
202-
try {
203-
await archiveStore.put(sample.key, "", {
204-
metadata: sample.metadata
205-
});
206-
result.archive_ok++;
207-
try {
208-
await sampleStore.delete(sample.key);
209-
result.delete_ok++;
210-
} catch (e) {
211-
console.log(`Delete failed. ${e}`);
212-
result.delete_fail++;
213-
}
214-
} catch (e) {
215-
console.log(`Archive failed. ${e}`);
216-
result.archive_fail++;
217-
result.delete_skip++;
218-
}
198+
cleanupStmts.push(context.env.DB
199+
.prepare("INSERT INTO sample_archive (time, data) VALUES (?, ?)")
200+
.bind(now, JSON.stringify(sample)));
201+
cleanupStmts.push(context.env.DB
202+
.prepare("DELETE FROM samples WHERE hash == ?")
203+
.bind(sample.hash));
219204
}
220-
}));
205+
});
206+
if (cleanupStmts.length > 0)
207+
await context.env.DB.batch(cleanupStmts);
221208

222209
return Response.json(result);
223210
}

‎functions/db-migrate.js‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
async function migrateArchive(context, result) {
2+
const now = Date.now();
3+
const archived = await context.env.ARCHIVE.list();
4+
const insertStmts = [];
5+
const keysToDelete = [];
6+
7+
// Limit batch size to stay within request limits.
8+
for (const k of archived.keys) {
9+
if (insertStmts.length >= 500) {
10+
result.archive_has_more = true;
11+
break;
12+
}
13+
14+
const metadata = k.metadata;
15+
metadata.hash = k.name;
16+
insertStmts.push(context.env.DB
17+
.prepare("INSERT INTO sample_archive (time, data) VALUES (?, ?)")
18+
.bind(now, JSON.stringify(metadata)));
19+
keysToDelete.push(k.name);
20+
}
21+
22+
if (insertStmts.length > 0) {
23+
await context.env.DB.batch(insertStmts);
24+
for (const k of keysToDelete) {
25+
await context.env.ARCHIVE.delete(k);
26+
}
27+
}
28+
29+
result.archive_insert_time = now;
30+
result.archive_migrated = keysToDelete.length;
31+
}
32+
33+
async function migrateSamples(context, result) {
34+
const now = Date.now();
35+
const samples = await context.env.SAMPLES.list();
36+
const insertStmts = [];
37+
const keysToDelete = [];
38+
39+
// Limit batch size to stay within request limits.
40+
for (const k of samples.keys) {
41+
if (insertStmts.length >= 500) {
42+
result.samples_has_more = true;
43+
break;
44+
}
45+
46+
const metadata = k.metadata;
47+
insertStmts.push(context.env.DB
48+
.prepare(`
49+
INSERT OR IGNORE INTO samples
50+
(hash, time, rssi, snr, observed, repeaters)
51+
VALUES (?, ?, ?, ?, ?, ?)`)
52+
.bind(
53+
k.name,
54+
metadata.time,
55+
metadata.rssi ?? null,
56+
metadata.snr ?? null,
57+
metadata.observed ?? 0,
58+
JSON.stringify(metadata.path ?? [])
59+
));
60+
keysToDelete.push(k.name);
61+
}
62+
63+
if (insertStmts.length > 0) {
64+
await context.env.DB.batch(insertStmts);
65+
for (const k of keysToDelete) {
66+
await context.env.SAMPLES.delete(k);
67+
}
68+
}
69+
70+
result.samples_insert_time = now;
71+
result.samples_migrated = keysToDelete.length;
72+
}
73+
74+
export async function onRequest(context) {
75+
const result = {};
76+
const url = new URL(context.request.url);
77+
const op = url.searchParams.get('op');
78+
79+
switch (op) {
80+
case "archive":
81+
await migrateArchive(context, result);
82+
break;
83+
case "samples":
84+
await migrateSamples(context, result);
85+
break;
86+
}
87+
88+
return Response.json(result);
89+
}

‎functions/get-nodes.js‎

Lines changed: 18 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@ import * as util from '../content/shared.js';
44

55
export async function onRequest(context) {
66
const coverageStore = context.env.COVERAGE;
7-
const sampleStore = context.env.SAMPLES;
87
const repeaterStore = context.env.REPEATERS;
98
const responseData = {
109
coverage: [],
@@ -46,27 +45,26 @@ export async function onRequest(context) {
4645

4746
// Samples
4847
// TODO: merge samples into coverage server-side?
49-
do {
50-
const samplesList = await sampleStore.list({ cursor: cursor });
51-
cursor = samplesList.cursor ?? null;
52-
samplesList.keys.forEach(s => {
53-
const path = s.metadata.path ?? [];
54-
const item = {
55-
id: s.name,
56-
time: util.truncateTime(s.metadata.time ?? 0),
57-
obs: s.metadata.observed ?? path.length > 0
58-
};
48+
const { results: samples } = await context.env.DB
49+
.prepare("SELECT * FROM samples").run();
50+
console.log(samples);
51+
samples.forEach(s => {
52+
const path = JSON.parse(s.repeaters);
53+
const item = {
54+
id: s.hash,
55+
time: util.truncateTime(s.time ?? 0),
56+
obs: s.observed
57+
};
5958

60-
// Don't send empty values.
61-
if (path.length > 0) {
62-
item.path = path
63-
};
64-
if (s.metadata.snr) item.snr = s.metadata.snr;
65-
if (s.metadata.rssi) item.rssi = s.metadata.rssi;
59+
// Don't send empty values.
60+
if (path.length > 0) {
61+
item.path = path
62+
};
63+
if (s.snr != null) item.snr = s.snr;
64+
if (s.rssi != null) item.rssi = s.rssi;
6665

67-
responseData.samples.push(item);
68-
});
69-
} while (cursor !== null)
66+
responseData.samples.push(item);
67+
});
7068

7169
// Repeaters
7270
do {

0 commit comments

Comments
 (0)