Merge tracker/events store into a single indexeddb store
This commit is contained in:
parent
45cbd1d6ba
commit
8a0918d551
3 changed files with 64 additions and 84 deletions
|
|
@ -3,7 +3,7 @@ import type {Unsubscriber} from "svelte/store"
|
||||||
import {deleteDB} from "idb"
|
import {deleteDB} from "idb"
|
||||||
import {SecureStorage} from "@aparajita/capacitor-secure-storage"
|
import {SecureStorage} from "@aparajita/capacitor-secure-storage"
|
||||||
import {Preferences} from "@capacitor/preferences"
|
import {Preferences} from "@capacitor/preferences"
|
||||||
import {noop, on, throttle, batch, call, addToMapKey, makeQueue} from "@welshman/lib"
|
import {noop, on, throttle, batch, call, makeQueue} from "@welshman/lib"
|
||||||
import type {Maybe} from "@welshman/lib"
|
import type {Maybe} from "@welshman/lib"
|
||||||
import {
|
import {
|
||||||
ALERT_ANDROID,
|
ALERT_ANDROID,
|
||||||
|
|
@ -111,7 +111,6 @@ export const ss = call(() => {
|
||||||
|
|
||||||
const TABLES = [
|
const TABLES = [
|
||||||
{name: "events", keyPath: "id"},
|
{name: "events", keyPath: "id"},
|
||||||
{name: "tracker", keyPath: "id"},
|
|
||||||
{name: "relays", keyPath: "url"},
|
{name: "relays", keyPath: "url"},
|
||||||
{name: "relayStats", keyPath: "url"},
|
{name: "relayStats", keyPath: "url"},
|
||||||
{name: "handles", keyPath: "nip05"},
|
{name: "handles", keyPath: "nip05"},
|
||||||
|
|
@ -154,7 +153,7 @@ const isRelayScoped = (event: TrustedEvent) =>
|
||||||
const shouldPersistEvent = (event: TrustedEvent) =>
|
const shouldPersistEvent = (event: TrustedEvent) =>
|
||||||
kinds.meta.includes(event.kind) || kinds.alert.includes(event.kind) || isRelayScoped(event)
|
kinds.meta.includes(event.kind) || kinds.alert.includes(event.kind) || isRelayScoped(event)
|
||||||
|
|
||||||
type TrackerItem = {id: string; relays: string[]}
|
type EventItem = {id: string; event: TrustedEvent; relays: string[]}
|
||||||
|
|
||||||
type PlaintextItem = {key: string; value: string}
|
type PlaintextItem = {key: string; value: string}
|
||||||
|
|
||||||
|
|
@ -225,62 +224,53 @@ class Storage {
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Events load together with the relays they came from, since a relay-scoped event that lost
|
* Each row holds an event together with the relays it came from, since a relay-scoped event
|
||||||
* its provenance can never be keyed to a space again — it sits in the repository invisible,
|
* that lost its provenance can never be keyed to a space again — it sits in the repository
|
||||||
* and negentropy reconciles it away as already-present, so it never comes back. Drop those
|
* invisible, and negentropy reconciles it away as already-present, so it never comes back.
|
||||||
* and let the next sync pull them down again.
|
* Drop those and let the next sync pull them down again.
|
||||||
*/
|
*/
|
||||||
private loadCriticalData = async () => {
|
private loadCriticalData = async () => {
|
||||||
const eventsTable = this.db.table<TrustedEvent>("events")
|
const table = this.db.table<EventItem>("events")
|
||||||
const trackerTable = this.db.table<TrackerItem>("tracker")
|
const items: EventItem[] = []
|
||||||
const [storedEvents, storedItems] = await Promise.all([
|
const stale: string[] = []
|
||||||
eventsTable.getAll(),
|
|
||||||
trackerTable.getAll(),
|
|
||||||
])
|
|
||||||
|
|
||||||
const relaysById = new Map(storedItems.map(({id, relays}) => [id, new Set(relays)]))
|
for (const item of await table.getAll()) {
|
||||||
const events: TrustedEvent[] = []
|
// Rows written before provenance was stored inline hold a bare event
|
||||||
const staleEvents: string[] = []
|
if (
|
||||||
|
item.event &&
|
||||||
for (const event of storedEvents) {
|
shouldPersistEvent(item.event) &&
|
||||||
if (shouldPersistEvent(event) && (!isRelayScoped(event) || relaysById.has(event.id))) {
|
(!isRelayScoped(item.event) || item.relays.length > 0)
|
||||||
event[verifiedSymbol] = true
|
) {
|
||||||
events.push(event)
|
item.event[verifiedSymbol] = true
|
||||||
|
items.push(item)
|
||||||
} else {
|
} else {
|
||||||
staleEvents.push(event.id)
|
stale.push(item.id)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
for (const [id, relays] of this.app.tracker.relaysById) {
|
this.app.repository.load(items.map(item => item.event))
|
||||||
for (const relay of relays) {
|
|
||||||
addToMapKey(relaysById, id, relay)
|
const relaysById = new Map<string, Set<string>>()
|
||||||
|
|
||||||
|
for (const {id, relays} of items) {
|
||||||
|
// Anything the repository rejected was superseded by an event that arrived before we
|
||||||
|
// finished loading
|
||||||
|
if (this.app.repository.getEvent(id)) {
|
||||||
|
relaysById.set(id, new Set(relays))
|
||||||
|
} else {
|
||||||
|
stale.push(id)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
this.app.repository.load([...this.app.repository.dump(), ...events])
|
|
||||||
|
|
||||||
const staleItems = storedItems
|
|
||||||
.map(item => item.id)
|
|
||||||
.filter(id => !this.app.repository.getEvent(id))
|
|
||||||
|
|
||||||
for (const id of staleItems) {
|
|
||||||
relaysById.delete(id)
|
|
||||||
}
|
|
||||||
|
|
||||||
this.app.tracker.load(relaysById)
|
this.app.tracker.load(relaysById)
|
||||||
|
|
||||||
if (staleEvents.length > 0) {
|
if (stale.length > 0) {
|
||||||
void eventsTable.bulkDelete(staleEvents)
|
void table.bulkDelete(stale)
|
||||||
}
|
|
||||||
|
|
||||||
if (staleItems.length > 0) {
|
|
||||||
void trackerTable.bulkDelete(staleItems)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private syncEvents = () => {
|
private syncEvents = () => {
|
||||||
const eventsTable = this.db.table<TrustedEvent>("events")
|
const table = this.db.table<EventItem>("events")
|
||||||
const trackerTable = this.db.table<TrackerItem>("tracker")
|
|
||||||
|
|
||||||
return on(
|
return on(
|
||||||
this.app.repository,
|
this.app.repository,
|
||||||
|
|
@ -304,73 +294,55 @@ class Storage {
|
||||||
|
|
||||||
if (add.length > 0) {
|
if (add.length > 0) {
|
||||||
// The ingest policy tracks an event before publishing it, so by the time the
|
// The ingest policy tracks an event before publishing it, so by the time the
|
||||||
// repository reports it, its provenance is already in the tracker — persist the
|
// repository reports it, its provenance is already in the tracker
|
||||||
// two together so an event is never saved without the relays it came from.
|
await table.bulkPut(
|
||||||
const items: TrackerItem[] = []
|
add.map(event => ({
|
||||||
|
id: event.id,
|
||||||
for (const {id} of add) {
|
event,
|
||||||
const relays = Array.from(this.app.tracker.getRelays(id))
|
relays: Array.from(this.app.tracker.getRelays(event.id)),
|
||||||
|
})),
|
||||||
if (relays.length > 0) {
|
)
|
||||||
items.push({id, relays})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
await eventsTable.bulkPut(add)
|
|
||||||
await trackerTable.bulkPut(items)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if (remove.size > 0) {
|
if (remove.size > 0) {
|
||||||
await eventsTable.bulkDelete(remove)
|
await table.bulkDelete(remove)
|
||||||
await trackerTable.bulkDelete(remove)
|
|
||||||
}
|
}
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
private syncTracker = () => {
|
private syncTracker = () => {
|
||||||
const table = this.db.table<TrackerItem>("tracker")
|
const table = this.db.table<EventItem>("events")
|
||||||
|
|
||||||
const _onAdd = async (ids: Iterable<string>) => {
|
const _onChange = async (ids: Iterable<string>) => {
|
||||||
const items: TrackerItem[] = []
|
const items: EventItem[] = []
|
||||||
|
|
||||||
for (const id of ids) {
|
for (const id of ids) {
|
||||||
const event = this.app.repository.getEvent(id)
|
const event = this.app.repository.getEvent(id)
|
||||||
|
|
||||||
// A brand-new event is tracked before it's published, so it isn't queryable here —
|
// A brand-new event is tracked before it's published, so it isn't queryable here —
|
||||||
// syncEvents persists its provenance along with the event itself. This pass only
|
// syncEvents persists its provenance along with the event itself. This pass only
|
||||||
// records additional relays for events we already have.
|
// records relay changes for events we already have.
|
||||||
if (!event || !shouldPersistEvent(event)) continue
|
if (event && shouldPersistEvent(event)) {
|
||||||
|
items.push({id, event, relays: Array.from(this.app.tracker.getRelays(id))})
|
||||||
const relays = Array.from(this.app.tracker.getRelays(id))
|
}
|
||||||
|
|
||||||
if (relays.length === 0) continue
|
|
||||||
|
|
||||||
items.push({id, relays})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
await table.bulkPut(items)
|
await table.bulkPut(items)
|
||||||
}
|
}
|
||||||
|
|
||||||
const _onRemove = async (ids: Iterable<string>) => {
|
const onAdd = batch(3000, _onChange)
|
||||||
await table.bulkDelete(Array.from(ids))
|
const onRemove = batch(3000, _onChange)
|
||||||
}
|
const onLoad = () => _onChange(this.app.tracker.relaysById.keys())
|
||||||
|
|
||||||
const onAdd = batch(3000, _onAdd)
|
|
||||||
const onRemove = batch(3000, _onRemove)
|
|
||||||
const onLoad = () => _onAdd(this.app.tracker.relaysById.keys())
|
|
||||||
const onClear = () => _onRemove(this.app.tracker.relaysById.keys())
|
|
||||||
|
|
||||||
this.app.tracker.on("add", onAdd)
|
this.app.tracker.on("add", onAdd)
|
||||||
this.app.tracker.on("remove", onRemove)
|
this.app.tracker.on("remove", onRemove)
|
||||||
this.app.tracker.on("load", onLoad)
|
this.app.tracker.on("load", onLoad)
|
||||||
this.app.tracker.on("clear", onClear)
|
|
||||||
|
|
||||||
return () => {
|
return () => {
|
||||||
this.app.tracker.off("add", onAdd)
|
this.app.tracker.off("add", onAdd)
|
||||||
this.app.tracker.off("remove", onRemove)
|
this.app.tracker.off("remove", onRemove)
|
||||||
this.app.tracker.off("load", onLoad)
|
this.app.tracker.off("load", onLoad)
|
||||||
this.app.tracker.off("clear", onClear)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,6 @@
|
||||||
import {openDB, deleteDB} from "idb"
|
import {openDB, deleteDB} from "idb"
|
||||||
import type {IDBPDatabase} from "idb"
|
import type {IDBPDatabase} from "idb"
|
||||||
|
import {spec} from "@welshman/lib"
|
||||||
import type {Maybe} from "@welshman/lib"
|
import type {Maybe} from "@welshman/lib"
|
||||||
|
|
||||||
export type IDBStore = {
|
export type IDBStore = {
|
||||||
|
|
@ -19,14 +20,17 @@ export class IDB {
|
||||||
constructor(readonly options: IDBOptions) {}
|
constructor(readonly options: IDBOptions) {}
|
||||||
|
|
||||||
// Object stores can only be created during a version change, and which stores we need depends
|
// Object stores can only be created during a version change, and which stores we need depends
|
||||||
// on who is logged in, so open at whatever version exists and bump it to add missing ones.
|
// on who is logged in, so open at whatever version exists and bump it to reconcile the schema.
|
||||||
private open = async () => {
|
private open = async () => {
|
||||||
const {name, stores} = this.options
|
const {name, stores} = this.options
|
||||||
const blocking = () => this.close()
|
const blocking = () => this.close()
|
||||||
const db = await openDB(name, undefined, {blocking})
|
const db = await openDB(name, undefined, {blocking})
|
||||||
const missing = stores.filter(store => !db.objectStoreNames.contains(store.name))
|
const missing = stores.filter(store => !db.objectStoreNames.contains(store.name))
|
||||||
|
const obsolete = Array.from(db.objectStoreNames).filter(
|
||||||
|
storeName => !stores.some(spec({name: storeName})),
|
||||||
|
)
|
||||||
|
|
||||||
if (missing.length === 0) {
|
if (missing.length === 0 && obsolete.length === 0) {
|
||||||
return db
|
return db
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -39,6 +43,10 @@ export class IDB {
|
||||||
for (const {name, keyPath} of missing) {
|
for (const {name, keyPath} of missing) {
|
||||||
idbDb.createObjectStore(name, {keyPath})
|
idbDb.createObjectStore(name, {keyPath})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
for (const storeName of obsolete) {
|
||||||
|
idbDb.deleteObjectStore(storeName)
|
||||||
|
}
|
||||||
},
|
},
|
||||||
blocked: (currentVersion, blockedVersion) =>
|
blocked: (currentVersion, blockedVersion) =>
|
||||||
console.error(
|
console.error(
|
||||||
|
|
|
||||||
|
|
@ -213,8 +213,8 @@
|
||||||
unsubscribers.push(Push.sync())
|
unsubscribers.push(Push.sync())
|
||||||
|
|
||||||
// Logging in swaps in a fresh app — its policies rebind themselves, we just have to sync
|
// Logging in swaps in a fresh app — its policies rebind themselves, we just have to sync
|
||||||
// application data against the new identity's relays. Loading the new app's storage
|
// application data against the new identity's relays. Wait for the new app's storage to
|
||||||
// replaces its repository and tracker wholesale, so wait for it the way startup does.
|
// load the way startup does, so sync reconciles against cached events.
|
||||||
let currentApp = core.app.get()
|
let currentApp = core.app.get()
|
||||||
|
|
||||||
const resync = async () => {
|
const resync = async () => {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue