forked from coracle/flotilla
Speed up boot, prune stores
This commit is contained in:
Jon Staab
+13
-19
@@ -1,6 +1,5 @@
|
||||
import {call} from "@welshman/lib"
|
||||
import {Preferences} from "@capacitor/preferences"
|
||||
import {Filesystem, Directory} from "@capacitor/filesystem"
|
||||
import {IDB} from "@lib/indexeddb"
|
||||
|
||||
export const kv = call(() => {
|
||||
@@ -31,22 +30,17 @@ export const kv = call(() => {
|
||||
return {get, set, clear}
|
||||
})
|
||||
|
||||
export const db = new IDB({name: "flotilla-9gl", version: 1})
|
||||
|
||||
// Migration - we used to use capacitor's filesystem for storage, clear it out since we're
|
||||
// going back to indexeddb
|
||||
call(async () => {
|
||||
const res = await Filesystem.readdir({
|
||||
path: "",
|
||||
directory: Directory.Data,
|
||||
})
|
||||
|
||||
await Promise.all(
|
||||
res.files.map(file =>
|
||||
Filesystem.deleteFile({
|
||||
path: file.name,
|
||||
directory: Directory.Data,
|
||||
}),
|
||||
),
|
||||
)
|
||||
export const db = new IDB({
|
||||
name: "flotilla-9gl",
|
||||
version: 1,
|
||||
stores: [
|
||||
{name: "events", keyPath: "id"},
|
||||
{name: "tracker", keyPath: "id"},
|
||||
{name: "relays", keyPath: "url"},
|
||||
{name: "relayStats", keyPath: "url"},
|
||||
{name: "handles", keyPath: "nip05"},
|
||||
{name: "zappers", keyPath: "lnurl"},
|
||||
{name: "plaintext", keyPath: "key"},
|
||||
{name: "wrapManager", keyPath: "id"},
|
||||
],
|
||||
})
|
||||
|
||||
+254
-193
@@ -45,9 +45,8 @@ import {
|
||||
wrapManager,
|
||||
onRelay,
|
||||
} from "@welshman/app"
|
||||
import {isMobile} from "@lib/html"
|
||||
import type {IDBTable} from "@lib/indexeddb"
|
||||
import {MESSAGE_KINDS, DM_KINDS} from "@app/core/state"
|
||||
import type {Unsubscriber} from "svelte/store"
|
||||
import {db} from "@app/core/storage"
|
||||
|
||||
const kinds = {
|
||||
meta: [PROFILE, FOLLOWS, MUTES, RELAYS, BLOSSOM_SERVERS, MESSAGING_RELAYS, APP_DATA, ROOMS],
|
||||
@@ -62,204 +61,266 @@ const kinds = {
|
||||
ROOM_REMOVE_MEMBER,
|
||||
ROOM_CREATE_PERMISSION,
|
||||
],
|
||||
content: [...MESSAGE_KINDS, ...DM_KINDS],
|
||||
}
|
||||
|
||||
const rankEvent = (event: TrustedEvent) => {
|
||||
if (kinds.meta.includes(event.kind)) return 9
|
||||
if (kinds.alert.includes(event.kind)) return 8
|
||||
if (kinds.space.includes(event.kind)) return 7
|
||||
if (kinds.room.includes(event.kind)) return 6
|
||||
if (!isMobile && kinds.content.includes(event.kind)) return 5
|
||||
return 0
|
||||
}
|
||||
|
||||
const eventsAdapter = {
|
||||
name: "events",
|
||||
keyPath: "id",
|
||||
init: async (table: IDBTable<TrustedEvent>) => {
|
||||
const initialEvents = await table.getAll()
|
||||
|
||||
// Mark events verified to avoid re-verification of signatures
|
||||
for (const event of initialEvents) {
|
||||
event[verifiedSymbol] = true
|
||||
}
|
||||
|
||||
repository.load(initialEvents)
|
||||
|
||||
return on(
|
||||
repository,
|
||||
"update",
|
||||
batch(3000, async (updates: RepositoryUpdate[]) => {
|
||||
const add: TrustedEvent[] = []
|
||||
const remove = new Set<string>()
|
||||
|
||||
for (const update of updates) {
|
||||
for (const event of update.added) {
|
||||
if (rankEvent(event) > 0) {
|
||||
add.push(event)
|
||||
remove.delete(event.id)
|
||||
}
|
||||
}
|
||||
|
||||
for (const id of update.removed) {
|
||||
remove.add(id)
|
||||
}
|
||||
}
|
||||
|
||||
if (add.length > 0) {
|
||||
await table.bulkPut(add)
|
||||
}
|
||||
|
||||
if (remove.size > 0) {
|
||||
await table.bulkDelete(remove)
|
||||
}
|
||||
}),
|
||||
)
|
||||
},
|
||||
}
|
||||
const shouldPersistEvent = (event: TrustedEvent) =>
|
||||
kinds.meta.includes(event.kind) ||
|
||||
kinds.alert.includes(event.kind) ||
|
||||
kinds.space.includes(event.kind) ||
|
||||
kinds.room.includes(event.kind)
|
||||
|
||||
type TrackerItem = {id: string; relays: string[]}
|
||||
|
||||
const trackerAdapter = {
|
||||
name: "tracker",
|
||||
keyPath: "id",
|
||||
init: async (table: IDBTable<TrackerItem>) => {
|
||||
const relaysById = new Map<string, Set<string>>()
|
||||
|
||||
for (const {id, relays} of await table.getAll()) {
|
||||
relaysById.set(id, new Set(relays))
|
||||
}
|
||||
|
||||
tracker.load(relaysById)
|
||||
|
||||
const _onAdd = async (ids: Iterable<string>) => {
|
||||
const items: TrackerItem[] = []
|
||||
|
||||
for (const id of ids) {
|
||||
const event = repository.getEvent(id)
|
||||
|
||||
if (!event || rankEvent(event) === 0) continue
|
||||
|
||||
const relays = Array.from(tracker.getRelays(id))
|
||||
|
||||
if (relays.length === 0) continue
|
||||
|
||||
items.push({id, relays})
|
||||
}
|
||||
|
||||
await table.bulkPut(items)
|
||||
}
|
||||
|
||||
const _onRemove = async (ids: Iterable<string>) => {
|
||||
await table.bulkDelete(Array.from(ids))
|
||||
}
|
||||
|
||||
const onAdd = batch(3000, _onAdd)
|
||||
|
||||
const onRemove = batch(3000, _onRemove)
|
||||
|
||||
const onLoad = () => _onAdd(tracker.relaysById.keys())
|
||||
|
||||
const onClear = () => _onRemove(tracker.relaysById.keys())
|
||||
|
||||
tracker.on("add", onAdd)
|
||||
tracker.on("remove", onRemove)
|
||||
tracker.on("load", onLoad)
|
||||
tracker.on("clear", onClear)
|
||||
|
||||
return () => {
|
||||
tracker.off("add", onAdd)
|
||||
tracker.off("remove", onRemove)
|
||||
tracker.off("load", onLoad)
|
||||
tracker.off("clear", onClear)
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
const relaysAdapter = {
|
||||
name: "relays",
|
||||
keyPath: "url",
|
||||
init: async (table: IDBTable<RelayProfile>) => {
|
||||
relaysByUrl.set(indexBy(r => r.url, await table.getAll()))
|
||||
|
||||
return onRelay(batch(1000, table.bulkPut))
|
||||
},
|
||||
}
|
||||
|
||||
const relayStatsAdapter = {
|
||||
name: "relayStats",
|
||||
keyPath: "url",
|
||||
init: async (table: IDBTable<RelayStats>) => {
|
||||
relayStatsByUrl.set(indexBy(r => r.url, await table.getAll()))
|
||||
|
||||
return onRelayStats(batch(1000, table.bulkPut))
|
||||
},
|
||||
}
|
||||
|
||||
const handlesAdapter = {
|
||||
name: "handles",
|
||||
keyPath: "nip05",
|
||||
init: async (table: IDBTable<Handle>) => {
|
||||
handlesByNip05.set(indexBy(r => r.nip05, await table.getAll()))
|
||||
|
||||
return onHandle(batch(1000, table.bulkPut))
|
||||
},
|
||||
}
|
||||
|
||||
const zappersAdapter = {
|
||||
name: "zappers",
|
||||
keyPath: "lnurl",
|
||||
init: async (table: IDBTable<Zapper>) => {
|
||||
zappersByLnurl.set(indexBy(z => z.lnurl, await table.getAll()))
|
||||
|
||||
return onZapper(batch(3000, table.bulkPut))
|
||||
},
|
||||
}
|
||||
|
||||
type PlaintextItem = {key: string; value: string}
|
||||
|
||||
const plaintextAdapter = {
|
||||
name: "plaintext",
|
||||
keyPath: "key",
|
||||
init: async (table: IDBTable<PlaintextItem>) => {
|
||||
const initialRecords = await table.getAll()
|
||||
const loadCriticalEvents = async () => {
|
||||
const table = db.table<TrustedEvent>("events")
|
||||
const initialEvents = await table.getAll()
|
||||
const keep: TrustedEvent[] = []
|
||||
const drop: string[] = []
|
||||
|
||||
plaintext.set(fromPairs(initialRecords.map(({key, value}) => [key, value])))
|
||||
|
||||
return throttled(3000, plaintext).subscribe($plaintext => {
|
||||
table.bulkPut(Object.entries($plaintext).map(([key, value]) => ({key, value})))
|
||||
})
|
||||
},
|
||||
}
|
||||
|
||||
const wrapManagerAdapter = {
|
||||
name: "wrapManager",
|
||||
keyPath: "id",
|
||||
init: async (table: IDBTable<WrapItem>) => {
|
||||
wrapManager.load(await table.getAll())
|
||||
|
||||
const addOne = batch(3000, table.bulkPut)
|
||||
|
||||
const removeOne = throttle(3000, table.bulkDelete)
|
||||
|
||||
wrapManager.on("add", addOne)
|
||||
wrapManager.on("remove", removeOne)
|
||||
|
||||
return () => {
|
||||
wrapManager.off("add", addOne)
|
||||
wrapManager.off("remove", removeOne)
|
||||
for (const event of initialEvents) {
|
||||
if (shouldPersistEvent(event)) {
|
||||
event[verifiedSymbol] = true
|
||||
keep.push(event)
|
||||
} else {
|
||||
drop.push(event.id)
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
repository.load(keep)
|
||||
|
||||
if (drop.length > 0) {
|
||||
void table.bulkDelete(drop)
|
||||
}
|
||||
}
|
||||
|
||||
export const adapters = [
|
||||
eventsAdapter,
|
||||
trackerAdapter,
|
||||
relaysAdapter,
|
||||
relayStatsAdapter,
|
||||
handlesAdapter,
|
||||
zappersAdapter,
|
||||
plaintextAdapter,
|
||||
wrapManagerAdapter,
|
||||
]
|
||||
const syncEvents = () => {
|
||||
const table = db.table<TrustedEvent>("events")
|
||||
|
||||
return on(
|
||||
repository,
|
||||
"update",
|
||||
batch(3000, async (updates: RepositoryUpdate[]) => {
|
||||
const add: TrustedEvent[] = []
|
||||
const remove = new Set<string>()
|
||||
|
||||
for (const update of updates) {
|
||||
for (const event of update.added) {
|
||||
if (shouldPersistEvent(event)) {
|
||||
add.push(event)
|
||||
remove.delete(event.id)
|
||||
}
|
||||
}
|
||||
|
||||
for (const id of update.removed) {
|
||||
remove.add(id)
|
||||
}
|
||||
}
|
||||
|
||||
if (add.length > 0) {
|
||||
await table.bulkPut(add)
|
||||
}
|
||||
|
||||
if (remove.size > 0) {
|
||||
await table.bulkDelete(remove)
|
||||
}
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
const loadCriticalTracker = async () => {
|
||||
const table = db.table<TrackerItem>("tracker")
|
||||
const relaysById = new Map<string, Set<string>>()
|
||||
const stale: string[] = []
|
||||
|
||||
for (const {id, relays} of await table.getAll()) {
|
||||
if (!repository.getEvent(id)) {
|
||||
stale.push(id)
|
||||
continue
|
||||
}
|
||||
|
||||
relaysById.set(id, new Set(relays))
|
||||
}
|
||||
|
||||
tracker.load(relaysById)
|
||||
|
||||
if (stale.length > 0) {
|
||||
void table.bulkDelete(stale)
|
||||
}
|
||||
}
|
||||
|
||||
const syncTracker = () => {
|
||||
const table = db.table<TrackerItem>("tracker")
|
||||
|
||||
const _onAdd = async (ids: Iterable<string>) => {
|
||||
const items: TrackerItem[] = []
|
||||
|
||||
for (const id of ids) {
|
||||
const event = repository.getEvent(id)
|
||||
|
||||
if (!event || !shouldPersistEvent(event)) continue
|
||||
|
||||
const relays = Array.from(tracker.getRelays(id))
|
||||
|
||||
if (relays.length === 0) continue
|
||||
|
||||
items.push({id, relays})
|
||||
}
|
||||
|
||||
await table.bulkPut(items)
|
||||
}
|
||||
|
||||
const _onRemove = async (ids: Iterable<string>) => {
|
||||
await table.bulkDelete(Array.from(ids))
|
||||
}
|
||||
|
||||
const onAdd = batch(3000, _onAdd)
|
||||
const onRemove = batch(3000, _onRemove)
|
||||
const onLoad = () => _onAdd(tracker.relaysById.keys())
|
||||
const onClear = () => _onRemove(tracker.relaysById.keys())
|
||||
|
||||
tracker.on("add", onAdd)
|
||||
tracker.on("remove", onRemove)
|
||||
tracker.on("load", onLoad)
|
||||
tracker.on("clear", onClear)
|
||||
|
||||
return () => {
|
||||
tracker.off("add", onAdd)
|
||||
tracker.off("remove", onRemove)
|
||||
tracker.off("load", onLoad)
|
||||
tracker.off("clear", onClear)
|
||||
}
|
||||
}
|
||||
|
||||
const loadCriticalRelays = async () => {
|
||||
const table = db.table<RelayProfile>("relays")
|
||||
|
||||
relaysByUrl.set(indexBy(r => r.url, await table.getAll()))
|
||||
}
|
||||
|
||||
const syncRelays = () => onRelay(batch(1000, db.table<RelayProfile>("relays").bulkPut))
|
||||
|
||||
const initRelayStats = async () => {
|
||||
const table = db.table<RelayStats>("relayStats")
|
||||
|
||||
relayStatsByUrl.set(indexBy(r => r.url, await table.getAll()))
|
||||
|
||||
return onRelayStats(batch(1000, table.bulkPut))
|
||||
}
|
||||
|
||||
const initHandles = async () => {
|
||||
const table = db.table<Handle>("handles")
|
||||
|
||||
handlesByNip05.set(indexBy(r => r.nip05, await table.getAll()))
|
||||
|
||||
return onHandle(batch(1000, table.bulkPut))
|
||||
}
|
||||
|
||||
const initZappers = async () => {
|
||||
const table = db.table<Zapper>("zappers")
|
||||
|
||||
zappersByLnurl.set(indexBy(z => z.lnurl, await table.getAll()))
|
||||
|
||||
return onZapper(batch(3000, table.bulkPut))
|
||||
}
|
||||
|
||||
const initPlaintext = async () => {
|
||||
const table = db.table<PlaintextItem>("plaintext")
|
||||
const initialRecords = await table.getAll()
|
||||
|
||||
plaintext.set(fromPairs(initialRecords.map(({key, value}) => [key, value])))
|
||||
|
||||
return throttled(3000, plaintext).subscribe($plaintext => {
|
||||
table.bulkPut(Object.entries($plaintext).map(([key, value]) => ({key, value})))
|
||||
})
|
||||
}
|
||||
|
||||
const initWrapManager = async () => {
|
||||
const table = db.table<WrapItem>("wrapManager")
|
||||
|
||||
wrapManager.load(await table.getAll())
|
||||
|
||||
const addOne = batch(3000, table.bulkPut)
|
||||
const removeOne = throttle(3000, table.bulkDelete)
|
||||
|
||||
wrapManager.on("add", addOne)
|
||||
wrapManager.on("remove", removeOne)
|
||||
|
||||
return () => {
|
||||
wrapManager.off("add", addOne)
|
||||
wrapManager.off("remove", removeOne)
|
||||
}
|
||||
}
|
||||
|
||||
type StorageSync = {
|
||||
unsubscribe: Unsubscriber
|
||||
ready: Promise<void>
|
||||
}
|
||||
|
||||
export const sync = (): StorageSync => {
|
||||
const unsubscribers: Unsubscriber[] = []
|
||||
const deferredTimers: ReturnType<typeof setTimeout>[] = []
|
||||
let stopped = false
|
||||
|
||||
const addUnsubscriber = (unsubscriber: Unsubscriber) => {
|
||||
if (stopped) {
|
||||
unsubscriber()
|
||||
} else {
|
||||
unsubscribers.push(unsubscriber)
|
||||
}
|
||||
}
|
||||
|
||||
const scheduleDeferred = (task: () => Promise<void>) => {
|
||||
const timeout = setTimeout(() => {
|
||||
if (stopped) return
|
||||
|
||||
void task()
|
||||
}, 0)
|
||||
|
||||
deferredTimers.push(timeout)
|
||||
}
|
||||
|
||||
const ready = (async () => {
|
||||
await db.connect()
|
||||
|
||||
await Promise.all([loadCriticalEvents(), loadCriticalRelays()])
|
||||
await loadCriticalTracker()
|
||||
|
||||
addUnsubscriber(syncEvents())
|
||||
addUnsubscriber(syncTracker())
|
||||
addUnsubscriber(syncRelays())
|
||||
|
||||
scheduleDeferred(async () => {
|
||||
addUnsubscriber(await initRelayStats())
|
||||
})
|
||||
|
||||
scheduleDeferred(async () => {
|
||||
addUnsubscriber(await initHandles())
|
||||
})
|
||||
|
||||
scheduleDeferred(async () => {
|
||||
addUnsubscriber(await initZappers())
|
||||
})
|
||||
|
||||
scheduleDeferred(async () => {
|
||||
addUnsubscriber(await initPlaintext())
|
||||
})
|
||||
|
||||
scheduleDeferred(async () => {
|
||||
addUnsubscriber(await initWrapManager())
|
||||
})
|
||||
})()
|
||||
|
||||
const unsubscribe = () => {
|
||||
stopped = true
|
||||
|
||||
for (const timeout of deferredTimers) {
|
||||
clearTimeout(timeout)
|
||||
}
|
||||
|
||||
unsubscribers.forEach(unsubscriber => unsubscriber())
|
||||
}
|
||||
|
||||
return {unsubscribe, ready}
|
||||
}
|
||||
|
||||
+5
-19
@@ -1,39 +1,32 @@
|
||||
import {openDB, deleteDB} from "idb"
|
||||
import type {IDBPDatabase} from "idb"
|
||||
import type {Unsubscriber} from "svelte/store"
|
||||
import {call} from "@welshman/lib"
|
||||
import type {Maybe} from "@welshman/lib"
|
||||
|
||||
export type IDBAdapter = {
|
||||
export type IDBStore = {
|
||||
name: string
|
||||
keyPath: string
|
||||
init: (table: IDBTable<any>) => Promise<Unsubscriber>
|
||||
}
|
||||
|
||||
export type IDBAdapters = IDBAdapter[]
|
||||
|
||||
export type IDBOptions = {
|
||||
name: string
|
||||
version: number
|
||||
stores: IDBStore[]
|
||||
}
|
||||
|
||||
export class IDB {
|
||||
adapters: IDBAdapters = []
|
||||
connection: Maybe<Promise<IDBPDatabase>>
|
||||
unsubscribers: Maybe<Unsubscriber[]>
|
||||
failedToConnect = false
|
||||
|
||||
constructor(readonly options: IDBOptions) {}
|
||||
|
||||
async connect() {
|
||||
if (!this.failedToConnect && !this.connection) {
|
||||
const {name, version} = this.options
|
||||
const adapters = this.adapters
|
||||
const {name, version, stores} = this.options
|
||||
|
||||
try {
|
||||
this.connection = openDB(name, version, {
|
||||
upgrade(idbDb: IDBPDatabase) {
|
||||
const names = new Set(adapters.map(a => a.name))
|
||||
const names = new Set(stores.map(store => store.name))
|
||||
|
||||
for (const table of idbDb.objectStoreNames) {
|
||||
if (!names.has(table)) {
|
||||
@@ -41,7 +34,7 @@ export class IDB {
|
||||
}
|
||||
}
|
||||
|
||||
for (const {name, keyPath} of adapters) {
|
||||
for (const {name, keyPath} of stores) {
|
||||
try {
|
||||
idbDb.createObjectStore(name, {keyPath})
|
||||
} catch (e) {
|
||||
@@ -52,10 +45,6 @@ export class IDB {
|
||||
blocked() {},
|
||||
blocking() {},
|
||||
})
|
||||
|
||||
this.unsubscribers = await Promise.all(
|
||||
adapters.map(({name, init}) => init(this.table(name))),
|
||||
)
|
||||
} catch (e) {
|
||||
console.error("Failed to connect to indexeddb", e)
|
||||
this.failedToConnect = true
|
||||
@@ -115,9 +104,6 @@ export class IDB {
|
||||
}
|
||||
|
||||
close = () => {
|
||||
this.unsubscribers?.forEach(call)
|
||||
this.unsubscribers = undefined
|
||||
|
||||
this.connection?.then(c => c.close())
|
||||
this.connection = undefined
|
||||
}
|
||||
|
||||
@@ -126,11 +126,12 @@
|
||||
}),
|
||||
])
|
||||
|
||||
// Set up our storage adapters
|
||||
db.adapters = storage.adapters
|
||||
const storageSync = storage.sync()
|
||||
|
||||
// Wait until data storage is initialized before syncing other stuff
|
||||
await db.connect()
|
||||
unsubscribers.push(storageSync.unsubscribe)
|
||||
|
||||
// Wait for critical storage data only
|
||||
await storageSync.ready
|
||||
|
||||
// Close the database connection on reload
|
||||
unsubscribers.push(() => db.close())
|
||||
|
||||
Reference in New Issue
Block a user