implement batched fetch and fix adding adapter

This commit is contained in:
Iris Lightshard 2024-08-03 10:52:33 -06:00
parent 0106b445b5
commit 557b87b700
Signed by: Iris Lightshard
GPG key ID: 688407174966CAF3
11 changed files with 130 additions and 96 deletions

View file

@ -8,7 +8,7 @@ type Adapter interface {
Init(Settings, chan SocketData) error
Name() string
Subscribe(string) []error
Fetch(string, string) error
Fetch(string, []string) error
Do(string) error
DefaultSubscriptionFilter() string
}

View file

@ -90,7 +90,7 @@ func (self *MastoAdapter) Subscribe(filter string) []error {
return nil
}
func (self *MastoAdapter) Fetch(etype string, id string) error {
func (self *MastoAdapter) Fetch(etype string, ids []string) error {
return nil
}

View file

@ -265,7 +265,8 @@ func (self *MisskeyAdapter) toAuthor(usr mkm.User) *Author {
return &author
}
func (self *MisskeyAdapter) Fetch(etype, id string) error {
func (self *MisskeyAdapter) Fetch(etype, ids []string) error {
for _, id := range ids {
switch etype {
case "message":
data, err := self.mk.Notes().Show(id)
@ -336,6 +337,7 @@ func (self *MisskeyAdapter) Fetch(etype, id string) error {
}
}
}
return nil
}

View file

@ -79,7 +79,7 @@ func (self *NostrAdapter) Subscribe(filter string) []error {
return nil
}
func (self *NostrAdapter) Fetch(etype, id string) error {
func (self *NostrAdapter) Fetch(etype, ids []string) error {
return nil
}

View file

@ -3,6 +3,7 @@ import util from "./util"
import { Message, Author } from "./message"
import { MessageThread } from "./thread"
import { AdapterState } from "./adapter"
import { BatchTimer } from "./batch-timer"
export class AdapterElement extends HTMLElement {
static observedAttributes = [ "data-latest", "data-view", "data-viewing" ]
@ -12,6 +13,14 @@ export class AdapterElement extends HTMLElement {
private _name: string = ""
private _viewing: string = "";
private _convoyBatchTimer = new BatchTimer((ids: string[])=>{
let url = `/api/adapters/${this._name}/fetch?entity_type=convoy`;
for (let id of ids) {
url += `&entity_id=${id}`;
}
util.authorizedFetch("GET", url, null)
});
// TODO: use visibility of the thread to organize into DMs and public threads
private _threads: MessageThread[] = [];
private _orphans: Message[] = [];
@ -29,7 +38,6 @@ export class AdapterElement extends HTMLElement {
}
attributeChangedCallback() {
console.log(`${this._name}.attributeChangedCallback: start`);
// set the viewing subject if it's changed
const viewing = this.getAttribute("data-viewing");
if (this._viewing != viewing && viewing != null) {
@ -68,13 +76,11 @@ export class AdapterElement extends HTMLElement {
// if latest changed, check if it's a message
const latest = this.getAttribute("data-latest");
console.log(`${this._name}.attributeChangedCallback: checking latest(${latest}) vs _latest${this._latest}`);
if (latest ?? "" != this._latest) {
console.log("latest changed")
this._latest = latest ?? "";
let datastore = AdapterState._instance.data.get(this._name);
if (!datastore) {
util.errMsg(this._name + " has no datastore!");
//util.errMsg(this._name + " has no datastore!");
return;
}
const latestMsg = datastore.messages.get(this._latest);
@ -157,8 +163,11 @@ export class AdapterElement extends HTMLElement {
}
updateIdxView(latest: string, rootId: string) {
const existingThread = this.querySelector(`underbbs-thread-summary[data-msg="${rootId}"]`);
const existingThread = document.querySelector(`underbbs-thread-summary[data-msg='${rootId}']`);
const thread = this._threads.find(t=>t.root.data.id == rootId);
console.log(`looking for thread ${rootId}`)
console.log(`- DOM object: ${existingThread}`);
console.log(`- in memory: ${thread}`);
if (existingThread && thread) {
console.log(`updating thread: ${thread.root.data.id} // ${thread.messageCount} NEW`)
existingThread.setAttribute("data-latest", `${thread.latest}`);
@ -273,11 +282,7 @@ export class AdapterElement extends HTMLElement {
if (this._orphans.filter(o=>o.id == msg.id).length == 0) {
this._orphans.push(msg);
if (msg.replyTo) {
// request the parent's data, which will try to adopt this orphan when it comes in
util.authorizedFetch(
"GET",
`/api/adapters/${this._name}/fetch?entity_type=message&entity_id=${msg.replyTo}`,
null);
this._convoyBatchTimer.queue(k, 2000);
}
}

View file

@ -4,12 +4,14 @@ export class AdapterData {
public directMessages: Map<string, Message>;
public messages: Map<string, Message>;
public profileCache: Map<string, Author>;
public convoyCache: Map<string, string>;
constructor(protocol: string) {
this.protocol = protocol;
this.messages = new Map<string, Message>();
this.directMessages = new Map<string, Message>();
this.profileCache = new Map<string, Author>();
this.convoyCache = new Map<string, string>();
}
}

View file

@ -0,0 +1,24 @@
export class BatchTimer {
private _batch: string[];
private _timer: number;
private _reqFn: (id: string[])=>void;
constructor(reqFn: (id: string[])=>void) {
this._batch = [];
this._timer = new Date().getTime();
this._reqFn = reqFn;
}
public queue(id: string, timeout: number){
this._timer = new Date().getTime() + timeout;
this._batch.push(id);
setTimeout(this.checkBatch, timeout);
}
private checkBatch() {
if ((new Date()).getTime() >= this._timer) {
this._reqFn(this._batch);
this._batch = [];
}
}
}

View file

@ -125,7 +125,7 @@ export class SettingsElement extends HTMLElement {
self._adapters.push(adapterdata.nickname);
localStorage.setItem("settings", JSON.stringify(settings));
self.showSettings(self);
self.showSettings(self)();
}
}
}

View file

@ -1,4 +1,5 @@
import { DatagramSocket } from './websocket'
import { BatchTimer } from './batch-timer'
function _(key: string, value: any | null | undefined = undefined): any | null {
const x = <any>window;

View file

@ -167,7 +167,7 @@ func apiAdapterFetch(next http.Handler, subscribers map[*Subscriber][]adapter.Ad
queryParams := req.URL.Query()
for _, a := range subscribers[s] {
if a.Name() == apiParams["adapter_id"] {
err := a.Fetch(queryParams["entity_type"][0], queryParams["entity_id"][0])
err := a.Fetch(queryParams["entity_type"][0], queryParams["entity_id"])
if err != nil {
fmt.Println(err.Error())
w.WriteHeader(http.StatusInternalServerError)