Simplify storage adapters
This commit is contained in:
@@ -1,5 +1,5 @@
|
|||||||
import {readable, derived, type Readable} from "svelte/store"
|
import {readable, derived, type Readable, type Subscriber} from "svelte/store"
|
||||||
import {indexBy, type Maybe, now} from "@welshman/lib"
|
import {indexBy, remove, type Maybe, now} from "@welshman/lib"
|
||||||
import {withGetter} from "@welshman/store"
|
import {withGetter} from "@welshman/store"
|
||||||
import {getFreshness, setFreshnessThrottled} from "./freshness.js"
|
import {getFreshness, setFreshnessThrottled} from "./freshness.js"
|
||||||
|
|
||||||
@@ -18,6 +18,8 @@ export const collection = <T, LoadArgs extends any[]>({
|
|||||||
const pending = new Map<string, Promise<Maybe<T>>>()
|
const pending = new Map<string, Promise<Maybe<T>>>()
|
||||||
const loadAttempts = new Map<string, number>()
|
const loadAttempts = new Map<string, number>()
|
||||||
|
|
||||||
|
let subscribers: Subscriber<T>[] = []
|
||||||
|
|
||||||
const loadItem = async (key: string, ...args: LoadArgs) => {
|
const loadItem = async (key: string, ...args: LoadArgs) => {
|
||||||
const stale = indexStore.get().get(key)
|
const stale = indexStore.get().get(key)
|
||||||
|
|
||||||
@@ -65,6 +67,10 @@ export const collection = <T, LoadArgs extends any[]>({
|
|||||||
|
|
||||||
if (fresh) {
|
if (fresh) {
|
||||||
loadAttempts.delete(key)
|
loadAttempts.delete(key)
|
||||||
|
|
||||||
|
for (const subscriber of subscribers) {
|
||||||
|
subscriber(fresh)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return fresh
|
return fresh
|
||||||
@@ -82,5 +88,13 @@ export const collection = <T, LoadArgs extends any[]>({
|
|||||||
return derived(indexStore, $index => $index.get(key))
|
return derived(indexStore, $index => $index.get(key))
|
||||||
}
|
}
|
||||||
|
|
||||||
return {indexStore, deriveItem, loadItem}
|
const onItem = (cb: Subscriber<T>) => {
|
||||||
|
subscribers.push(cb)
|
||||||
|
|
||||||
|
return () => {
|
||||||
|
subscribers = remove(cb, subscribers)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return {indexStore, deriveItem, loadItem, onItem}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -82,6 +82,7 @@ export const {
|
|||||||
indexStore: handlesByNip05,
|
indexStore: handlesByNip05,
|
||||||
deriveItem: deriveHandle,
|
deriveItem: deriveHandle,
|
||||||
loadItem: loadHandle,
|
loadItem: loadHandle,
|
||||||
|
onItem: onHandle,
|
||||||
} = collection({
|
} = collection({
|
||||||
name: "handles",
|
name: "handles",
|
||||||
store: handles,
|
store: handles,
|
||||||
|
|||||||
@@ -85,6 +85,7 @@ export const {
|
|||||||
indexStore: relaysByUrl,
|
indexStore: relaysByUrl,
|
||||||
deriveItem: deriveRelay,
|
deriveItem: deriveRelay,
|
||||||
loadItem: loadRelay,
|
loadItem: loadRelay,
|
||||||
|
onItem: onRelay,
|
||||||
} = collection({
|
} = collection({
|
||||||
name: "relays",
|
name: "relays",
|
||||||
store: relays,
|
store: relays,
|
||||||
|
|||||||
+8
-188
@@ -2,7 +2,7 @@ import {openDB, deleteDB} from "idb"
|
|||||||
import {IDBPDatabase} from "idb"
|
import {IDBPDatabase} from "idb"
|
||||||
import {writable} from "svelte/store"
|
import {writable} from "svelte/store"
|
||||||
import {Unsubscriber, Writable} from "svelte/store"
|
import {Unsubscriber, Writable} from "svelte/store"
|
||||||
import {indexBy, equals, throttle, fromPairs} from "@welshman/lib"
|
import {indexBy, call, equals, throttle, fromPairs} from "@welshman/lib"
|
||||||
import {TrustedEvent} from "@welshman/util"
|
import {TrustedEvent} from "@welshman/util"
|
||||||
import {Repository} from "@welshman/relay"
|
import {Repository} from "@welshman/relay"
|
||||||
import {Tracker} from "@welshman/net"
|
import {Tracker} from "@welshman/net"
|
||||||
@@ -15,8 +15,8 @@ export type StorageAdapterOptions = {
|
|||||||
|
|
||||||
export type StorageAdapter = {
|
export type StorageAdapter = {
|
||||||
keyPath: string
|
keyPath: string
|
||||||
store: Writable<any[]>
|
init: () => Promise<void>
|
||||||
options: StorageAdapterOptions
|
sync: () => Unsubscriber
|
||||||
}
|
}
|
||||||
|
|
||||||
export let db: IDBPDatabase | undefined
|
export let db: IDBPDatabase | undefined
|
||||||
@@ -60,41 +60,6 @@ export const bulkDelete = async (name: string, ids: string[]) => {
|
|||||||
await tx.done
|
await tx.done
|
||||||
}
|
}
|
||||||
|
|
||||||
export const initIndexedDbAdapter = async (name: string, adapter: StorageAdapter) => {
|
|
||||||
let prevRecords = await getAll(name)
|
|
||||||
|
|
||||||
adapter.store.set(prevRecords)
|
|
||||||
|
|
||||||
setTimeout(() => {
|
|
||||||
adapter.store.subscribe(async (currentRecords: any[]) => {
|
|
||||||
if (dead.get()) {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
const currentIds = new Set(currentRecords.map(item => item[adapter.keyPath]))
|
|
||||||
const removedRecords = prevRecords.filter(r => !currentIds.has(r[adapter.keyPath]))
|
|
||||||
|
|
||||||
const prevRecordsById = indexBy(item => item[adapter.keyPath], prevRecords)
|
|
||||||
const updatedRecords = currentRecords.filter(
|
|
||||||
r => !equals(r, prevRecordsById.get(r[adapter.keyPath])),
|
|
||||||
)
|
|
||||||
|
|
||||||
prevRecords = currentRecords
|
|
||||||
|
|
||||||
if (updatedRecords.length > 0) {
|
|
||||||
await bulkPut(name, updatedRecords)
|
|
||||||
}
|
|
||||||
|
|
||||||
if (removedRecords.length > 0) {
|
|
||||||
await bulkDelete(
|
|
||||||
name,
|
|
||||||
removedRecords.map(item => item[adapter.keyPath]),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}, adapter.options.throttle || 0)
|
|
||||||
}
|
|
||||||
|
|
||||||
export const initStorage = async (
|
export const initStorage = async (
|
||||||
name: string,
|
name: string,
|
||||||
version: number,
|
version: number,
|
||||||
@@ -128,9 +93,11 @@ export const initStorage = async (
|
|||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
await Promise.all(
|
await Promise.all(Object.values(adapters).map(adapter => adapter.init()))
|
||||||
Object.entries(adapters).map(([name, config]) => initIndexedDbAdapter(name, config)),
|
|
||||||
)
|
const unsubscribers = Object.values(adapters).map(adapter => adapter.sync())
|
||||||
|
|
||||||
|
return () => unsubscribers.forEach(call)
|
||||||
}
|
}
|
||||||
|
|
||||||
export const closeStorage = async () => {
|
export const closeStorage = async () => {
|
||||||
@@ -146,150 +113,3 @@ export const clearStorage = async () => {
|
|||||||
db = undefined // force initStorage to run again in tests
|
db = undefined // force initStorage to run again in tests
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
const migrate = (data: any[], options: StorageAdapterOptions) =>
|
|
||||||
options.migrate ? options.migrate(data) : data
|
|
||||||
|
|
||||||
export const storageAdapters = {
|
|
||||||
fromCollectionStore: <T>(
|
|
||||||
keyPath: string,
|
|
||||||
store: Writable<T[]>,
|
|
||||||
options: StorageAdapterOptions = {},
|
|
||||||
) => ({
|
|
||||||
options,
|
|
||||||
keyPath,
|
|
||||||
store: throttled(options.throttle || 0, store),
|
|
||||||
}),
|
|
||||||
fromObjectStore: <T>(
|
|
||||||
store: Writable<Record<string, T>>,
|
|
||||||
options: StorageAdapterOptions = {},
|
|
||||||
) => ({
|
|
||||||
options,
|
|
||||||
keyPath: "key",
|
|
||||||
store: adapter({
|
|
||||||
store: throttled(options.throttle || 0, store),
|
|
||||||
forward: (data: Record<string, T>) =>
|
|
||||||
migrate(
|
|
||||||
Object.entries(data).map(([key, value]) => ({key, value})),
|
|
||||||
options,
|
|
||||||
),
|
|
||||||
backward: (data: {key: string; value: T}[]) =>
|
|
||||||
fromPairs(data.map(({key, value}) => [key, value])),
|
|
||||||
}),
|
|
||||||
}),
|
|
||||||
fromMapStore: <T>(store: Writable<Map<string, T>>, options: StorageAdapterOptions = {}) => ({
|
|
||||||
options,
|
|
||||||
keyPath: "key",
|
|
||||||
store: adapter({
|
|
||||||
store: throttled(options.throttle || 0, store),
|
|
||||||
forward: (data: Map<string, T>) =>
|
|
||||||
migrate(
|
|
||||||
Array.from(data.entries()).map(([key, value]) => ({key, value})),
|
|
||||||
options,
|
|
||||||
),
|
|
||||||
backward: (data: {key: string; value: T}[]) =>
|
|
||||||
new Map(data.map(({key, value}) => [key, value])),
|
|
||||||
}),
|
|
||||||
}),
|
|
||||||
fromTracker: (tracker: Tracker, options: StorageAdapterOptions = {}) => ({
|
|
||||||
options,
|
|
||||||
keyPath: "key",
|
|
||||||
store: custom(
|
|
||||||
setter => {
|
|
||||||
let onUpdate = () =>
|
|
||||||
setter(
|
|
||||||
migrate(
|
|
||||||
Array.from(tracker.relaysById.entries()).map(([key, urls]) => ({
|
|
||||||
key,
|
|
||||||
value: Array.from(urls),
|
|
||||||
})),
|
|
||||||
options,
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
if (options.throttle) {
|
|
||||||
onUpdate = throttle(options.throttle, onUpdate)
|
|
||||||
}
|
|
||||||
|
|
||||||
onUpdate()
|
|
||||||
tracker.on("update", onUpdate)
|
|
||||||
|
|
||||||
return () => tracker.off("update", onUpdate)
|
|
||||||
},
|
|
||||||
{
|
|
||||||
set: (data: {key: string; value: string[]}[]) =>
|
|
||||||
tracker.load(new Map(data.map(({key, value}) => [key, new Set(value)]))),
|
|
||||||
},
|
|
||||||
),
|
|
||||||
}),
|
|
||||||
fromRepository: (repository: Repository, options: StorageAdapterOptions = {}) => ({
|
|
||||||
options,
|
|
||||||
keyPath: "id",
|
|
||||||
store: custom(
|
|
||||||
setter => {
|
|
||||||
let onUpdate = () => setter(migrate(repository.dump(), options))
|
|
||||||
|
|
||||||
if (options.throttle) {
|
|
||||||
onUpdate = throttle(options.throttle, onUpdate)
|
|
||||||
}
|
|
||||||
|
|
||||||
onUpdate()
|
|
||||||
repository.on("update", onUpdate)
|
|
||||||
|
|
||||||
return () => repository.off("update", onUpdate)
|
|
||||||
},
|
|
||||||
{
|
|
||||||
set: (events: TrustedEvent[]) => repository.load(events),
|
|
||||||
},
|
|
||||||
),
|
|
||||||
}),
|
|
||||||
fromRepositoryAndTracker: (
|
|
||||||
repository: Repository,
|
|
||||||
tracker: Tracker,
|
|
||||||
options: StorageAdapterOptions = {},
|
|
||||||
) => ({
|
|
||||||
options,
|
|
||||||
keyPath: "id",
|
|
||||||
store: custom(
|
|
||||||
setter => {
|
|
||||||
let onUpdate = () => {
|
|
||||||
const events = migrate(repository.dump(), options)
|
|
||||||
|
|
||||||
setter(
|
|
||||||
events.map(event => {
|
|
||||||
const relays = Array.from(tracker.getRelays(event.id))
|
|
||||||
|
|
||||||
return {id: event.id, event, relays}
|
|
||||||
}),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
if (options.throttle) {
|
|
||||||
onUpdate = throttle(options.throttle, onUpdate)
|
|
||||||
}
|
|
||||||
|
|
||||||
onUpdate()
|
|
||||||
tracker.on("update", onUpdate)
|
|
||||||
repository.on("update", onUpdate)
|
|
||||||
|
|
||||||
return () => {
|
|
||||||
tracker.off("update", onUpdate)
|
|
||||||
}
|
|
||||||
},
|
|
||||||
{
|
|
||||||
set: (items: {event: TrustedEvent; relays: string[]}[]) => {
|
|
||||||
const events: TrustedEvent[] = []
|
|
||||||
const relaysById = new Map<string, Set<string>>()
|
|
||||||
|
|
||||||
for (const {event, relays} of items) {
|
|
||||||
events.push(event)
|
|
||||||
relaysById.set(event.id, new Set(relays))
|
|
||||||
}
|
|
||||||
|
|
||||||
repository.load(events)
|
|
||||||
tracker.load(relaysById)
|
|
||||||
},
|
|
||||||
},
|
|
||||||
),
|
|
||||||
}),
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -58,6 +58,7 @@ export const {
|
|||||||
indexStore: zappersByLnurl,
|
indexStore: zappersByLnurl,
|
||||||
deriveItem: deriveZapper,
|
deriveItem: deriveZapper,
|
||||||
loadItem: loadZapper,
|
loadItem: loadZapper,
|
||||||
|
onItem: onZapper,
|
||||||
} = collection({
|
} = collection({
|
||||||
name: "zappers",
|
name: "zappers",
|
||||||
store: zappers,
|
store: zappers,
|
||||||
|
|||||||
Reference in New Issue
Block a user