chore: adjust sync method
Some checks reported errors
continuous-integration/drone/push Build encountered an error
Some checks reported errors
continuous-integration/drone/push Build encountered an error
This commit is contained in:
parent
de1504b4bf
commit
4185f117cb
@ -13,7 +13,7 @@ export function useNotificationsView() {
|
|||||||
rb.withOptions({
|
rb.withOptions({
|
||||||
leaveOpen: true,
|
leaveOpen: true,
|
||||||
});
|
});
|
||||||
rb.withFilter().kinds(kinds).tag("p", [publicKey]).limit(100);
|
rb.withFilter().kinds(kinds).tag("p", [publicKey]);
|
||||||
return rb;
|
return rb;
|
||||||
}
|
}
|
||||||
}, [publicKey]);
|
}, [publicKey]);
|
||||||
|
@ -2,7 +2,7 @@ import "./Notifications.css";
|
|||||||
|
|
||||||
import { unwrap } from "@snort/shared";
|
import { unwrap } from "@snort/shared";
|
||||||
import { NostrEvent, NostrLink, TaggedNostrEvent } from "@snort/system";
|
import { NostrEvent, NostrLink, TaggedNostrEvent } from "@snort/system";
|
||||||
import { lazy, Suspense, useEffect, useMemo } from "react";
|
import { lazy, Suspense, useEffect, useMemo, useState } from "react";
|
||||||
|
|
||||||
import { AutoLoadMore } from "@/Components/Event/LoadMore";
|
import { AutoLoadMore } from "@/Components/Event/LoadMore";
|
||||||
import PageSpinner from "@/Components/PageSpinner";
|
import PageSpinner from "@/Components/PageSpinner";
|
||||||
@ -19,6 +19,7 @@ export default function NotificationsPage({ onClick }: { onClick?: (link: NostrL
|
|||||||
const login = useLogin();
|
const login = useLogin();
|
||||||
const { isMuted } = useModeration();
|
const { isMuted } = useModeration();
|
||||||
const groupInterval = 3600 * 6;
|
const groupInterval = 3600 * 6;
|
||||||
|
const [limit, setLimit] = useState(100);
|
||||||
|
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
markNotificationsRead(login);
|
markNotificationsRead(login);
|
||||||
@ -32,14 +33,16 @@ export default function NotificationsPage({ onClick }: { onClick?: (link: NostrL
|
|||||||
};
|
};
|
||||||
|
|
||||||
const myNotifications = useMemo(() => {
|
const myNotifications = useMemo(() => {
|
||||||
return notifications.filter(a => !isMuted(a.pubkey) && a.tags.some(b => b[0] === "p" && b[1] === login.publicKey));
|
return notifications
|
||||||
}, [notifications, login.publicKey]);
|
.sort((a, b) => a.created_at > b.created_at ? -1 : 1)
|
||||||
|
.slice(0, limit)
|
||||||
|
.filter(a => !isMuted(a.pubkey) && a.tags.some(b => b[0] === "p" && b[1] === login.publicKey));
|
||||||
|
}, [notifications, login.publicKey, limit]);
|
||||||
|
|
||||||
const timeGrouped = useMemo(() => {
|
const timeGrouped = useMemo(() => {
|
||||||
return myNotifications.reduce((acc, v) => {
|
return myNotifications.reduce((acc, v) => {
|
||||||
const key = `${timeKey(v)}:${getNotificationContext(v as TaggedNostrEvent)?.encode(CONFIG.eventLinkPrefix)}:${
|
const key = `${timeKey(v)}:${getNotificationContext(v as TaggedNostrEvent)?.encode(CONFIG.eventLinkPrefix)}:${v.kind
|
||||||
v.kind
|
}`;
|
||||||
}`;
|
|
||||||
if (acc.has(key)) {
|
if (acc.has(key)) {
|
||||||
unwrap(acc.get(key)).push(v as TaggedNostrEvent);
|
unwrap(acc.get(key)).push(v as TaggedNostrEvent);
|
||||||
} else {
|
} else {
|
||||||
@ -54,13 +57,13 @@ export default function NotificationsPage({ onClick }: { onClick?: (link: NostrL
|
|||||||
<div className="main-content">
|
<div className="main-content">
|
||||||
{CONFIG.features.notificationGraph && (
|
{CONFIG.features.notificationGraph && (
|
||||||
<Suspense fallback={<PageSpinner />}>
|
<Suspense fallback={<PageSpinner />}>
|
||||||
<NotificationGraph evs={myNotifications} />
|
<NotificationGraph evs={notifications} />
|
||||||
</Suspense>
|
</Suspense>
|
||||||
)}
|
)}
|
||||||
{login.publicKey &&
|
{login.publicKey &&
|
||||||
[...timeGrouped.entries()].map(([k, g]) => <NotificationGroup key={k} evs={g} onClick={onClick} />)}
|
[...timeGrouped.entries()].map(([k, g]) => <NotificationGroup key={k} evs={g} onClick={onClick} />)}
|
||||||
|
|
||||||
<AutoLoadMore onClick={() => {}} />
|
<AutoLoadMore onClick={() => { setLimit(l => l + 100) }} />
|
||||||
</div>
|
</div>
|
||||||
</>
|
</>
|
||||||
);
|
);
|
||||||
|
@ -20,7 +20,7 @@ export default function SyncAccountTool() {
|
|||||||
const relays = Object.entries(myRelays)
|
const relays = Object.entries(myRelays)
|
||||||
.filter(([, v]) => v.write)
|
.filter(([, v]) => v.write)
|
||||||
.map(([k]) => k);
|
.map(([k]) => k);
|
||||||
const sync = new RangeSync(system);
|
const sync = RangeSync.forSystem(system);
|
||||||
sync.on("event", evs => {
|
sync.on("event", evs => {
|
||||||
setResults(r => [...r, ...evs]);
|
setResults(r => [...r, ...evs]);
|
||||||
});
|
});
|
||||||
|
@ -8,9 +8,11 @@ import { DefaultConnectTimeout } from "./const";
|
|||||||
import { NostrEvent, OkResponse, ReqCommand, ReqFilter, TaggedNostrEvent, u256 } from "./nostr";
|
import { NostrEvent, OkResponse, ReqCommand, ReqFilter, TaggedNostrEvent, u256 } from "./nostr";
|
||||||
import { RelayInfo } from "./relay-info";
|
import { RelayInfo } from "./relay-info";
|
||||||
import EventKind from "./event-kind";
|
import EventKind from "./event-kind";
|
||||||
import { EventExt } from "./event-ext";
|
import { EventExt, EventType } from "./event-ext";
|
||||||
import { NegentropyFlow } from "./negentropy/negentropy-flow";
|
import { NegentropyFlow } from "./negentropy/negentropy-flow";
|
||||||
import { ConnectionType, ConnectionTypeEvents } from "./connection-pool";
|
import { ConnectionType, ConnectionTypeEvents } from "./connection-pool";
|
||||||
|
import { RangeSync } from "./sync";
|
||||||
|
import { NoteCollection } from "./note-collection";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Relay settings
|
* Relay settings
|
||||||
@ -395,15 +397,36 @@ export class Connection extends EventEmitter<ConnectionTypeEvents> implements Co
|
|||||||
} else if (cmd[0] === "SYNC") {
|
} else if (cmd[0] === "SYNC") {
|
||||||
const [_, id, eventSet, ...filters] = cmd;
|
const [_, id, eventSet, ...filters] = cmd;
|
||||||
const lastResortSync = () => {
|
const lastResortSync = () => {
|
||||||
if (filters.some(a => a.since || a.until || a.ids)) {
|
const isReplacableSync = filters.every(a => a.kinds?.every(b => EventExt.getType(b) === EventType.Replaceable || EventExt.getType(b) === EventType.ParameterizedReplaceable) ?? false);
|
||||||
|
if (filters.some(a => a.since || a.until || a.ids || a.limit) || isReplacableSync) {
|
||||||
this.request(["REQ", id, ...filters], item.cb);
|
this.request(["REQ", id, ...filters], item.cb);
|
||||||
} else {
|
} else {
|
||||||
|
const rs = RangeSync.forFetcher(async (rb, cb) => {
|
||||||
|
return await new Promise((resolve, reject) => {
|
||||||
|
const results = new NoteCollection();
|
||||||
|
const f = rb.buildRaw();
|
||||||
|
this.on("event", (c, e) => {
|
||||||
|
if (rb.id === c) {
|
||||||
|
cb?.([e]);
|
||||||
|
results.add(e);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
this.on("eose", s => {
|
||||||
|
if (s === rb.id) {
|
||||||
|
resolve(results.takeSnapshot());
|
||||||
|
}
|
||||||
|
});
|
||||||
|
this.request(["REQ", rb.id, ...f], undefined);
|
||||||
|
});
|
||||||
|
});
|
||||||
const latest = eventSet.reduce((acc, v) => (acc = v.created_at > acc ? v.created_at : acc), 0);
|
const latest = eventSet.reduce((acc, v) => (acc = v.created_at > acc ? v.created_at : acc), 0);
|
||||||
const newFilters = filters.map(a => ({
|
rs.setStartPoint(latest + 1);
|
||||||
...a,
|
rs.on("event", ev => {
|
||||||
since: latest + 1,
|
ev.forEach(e => this.emit("event", id, e));
|
||||||
}));
|
});
|
||||||
this.request(["REQ", id, ...newFilters], item.cb);
|
for (const f of filters) {
|
||||||
|
rs.sync(f);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
if (this.info?.negentropy === "v1") {
|
if (this.info?.negentropy === "v1") {
|
||||||
|
@ -1,6 +1,7 @@
|
|||||||
import { unixNow } from "@snort/shared";
|
import { unixNow } from "@snort/shared";
|
||||||
import EventEmitter from "eventemitter3";
|
import EventEmitter from "eventemitter3";
|
||||||
import { ReqFilter, RequestBuilder, SystemInterface, TaggedNostrEvent } from "..";
|
import { ReqFilter, RequestBuilder, SystemInterface, TaggedNostrEvent } from "..";
|
||||||
|
import { v4 as uuid } from "uuid";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* When nostr was created
|
* When nostr was created
|
||||||
@ -16,13 +17,27 @@ interface RangeSyncEvents {
|
|||||||
* A simple time based sync for pulling lots of data from nostr
|
* A simple time based sync for pulling lots of data from nostr
|
||||||
*/
|
*/
|
||||||
export class RangeSync extends EventEmitter<RangeSyncEvents> {
|
export class RangeSync extends EventEmitter<RangeSyncEvents> {
|
||||||
|
#id = uuid();
|
||||||
#start: number = NostrBirthday;
|
#start: number = NostrBirthday;
|
||||||
#windowSize: number = 60 * 60 * 12;
|
#windowSize: number = 60 * 60 * 12;
|
||||||
|
#fetcher!: SystemInterface["Fetch"];
|
||||||
|
|
||||||
constructor(readonly system: SystemInterface) {
|
private constructor() {
|
||||||
super();
|
super();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static forSystem(system: SystemInterface) {
|
||||||
|
const rs = new RangeSync();
|
||||||
|
rs.#fetcher = (r, c) => system.Fetch(r, c);
|
||||||
|
return rs;
|
||||||
|
}
|
||||||
|
|
||||||
|
static forFetcher(fn: SystemInterface["Fetch"]) {
|
||||||
|
const rs = new RangeSync();
|
||||||
|
rs.#fetcher = fn;
|
||||||
|
return rs;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Set window size in seconds
|
* Set window size in seconds
|
||||||
*/
|
*/
|
||||||
@ -52,18 +67,15 @@ export class RangeSync extends EventEmitter<RangeSyncEvents> {
|
|||||||
throw new Error("Filter must not contain since/until/limit");
|
throw new Error("Filter must not contain since/until/limit");
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!this.system.requestRouter) {
|
|
||||||
throw new Error("RangeSync cannot work without request router!");
|
|
||||||
}
|
|
||||||
|
|
||||||
const now = unixNow();
|
const now = unixNow();
|
||||||
|
let ctr = 1;
|
||||||
for (let end = now; end > this.#start; end -= this.#windowSize) {
|
for (let end = now; end > this.#start; end -= this.#windowSize) {
|
||||||
const rb = new RequestBuilder(`range-query:${end}`);
|
const rb = new RequestBuilder(`${this.#id}+${ctr++}`);
|
||||||
rb.withBareFilter(filter)
|
rb.withBareFilter(filter)
|
||||||
.since(end - this.#windowSize)
|
.since(end - this.#windowSize)
|
||||||
.until(end);
|
.until(end);
|
||||||
this.emit("scan", end);
|
this.emit("scan", end);
|
||||||
const results = await this.system.Fetch(rb);
|
const results = await this.#fetcher(rb);
|
||||||
this.emit("event", results);
|
this.emit("event", results);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user