const e = require("ready-resource"),
t = require("hypercore-id-encoding"),
s = require("scope-lock"),
r = require("resolve-reject-promise"),
i = require("safety-catch"),
{
local: {
hyperschema: { getEnum: n },
},
} = require("../../../schema/spec"),
{ Task: a } = require("./task"),
o = n("@local/tasks"),
c = require("./rooms/leave"),
l = require("./rooms/clear"),
h = require("./rooms/create"),
u = require("./identity/restore"),
d = require("./pairing/invitation"),
k = require("./pairing/response"),
p = require("./pairing/keys"),
y = require("./contacts/request"),
m = require("./testing/fast"),
_ = require("./testing/slow"),
w = require("./testing/error"),
g = 10
class R extends e {
constructor(e, s, n) {
;(super(),
(this.description = n),
(this.manager = e),
(this.idx = 0),
(this.id = s),
(this.keet = this.manager.keet),
(this.scope = this.description.discoveryKey),
(this.scopeId = t.encode(this.scope)),
(this.running = null),
(this.waiting = !0),
(this.completed = !1),
(this._task = null))
const { resolve: a, reject: o, promise: c } = r()
;((this._resolve = a), (this._reject = o), (this._promise = c), this._promise.catch(i))
}
get taskClosing() {
return this._task && this._task.closing
}
async _close() {
;(this._task && (await this._task.close()), this._reject(new Error("Task was cancelled")))
}
async _run() {
if (this.manager.closing) return
if (
(await this.manager.internalFlushTasks(),
(this._task = this._createTask()),
await this.manager.internalHandleTaskStarted(this),
this.manager.closing)
)
return
this.waiting = !1
let e = null,
t = null
try {
e = await this._task.run()
} catch (e) {
if (this.manager.closing) return
t = e
} finally {
;((this.completed = !0),
await this.manager.internalHandleTaskCompleted(this),
t ? this._reject(t) : this._resolve(e))
}
}
_createTask() {
switch (this.description.type) {
case o.RestoreIdentity:
return new u(this)
case o.CreateRoom:
return new h(this)
case o.LeaveRoom:
return new c(this)
case o.ClearRoom:
return new l(this)
case o.JoinRoomInvitation:
return new d(this)
case o.JoinRoomKeys:
return new p(this)
case o.JoinRoomResponse:
return new k(this)
case o.SendContactRequest:
return new y(this)
case o.TestFastTask:
return new m(this)
case o.TestSlowTask:
return new _(this)
case o.TestErrorTask:
return new w(this)
default:
throw new Error("Unknown task type: " + this.description.type)
}
}
run() {
return (
this.running || ((this.running = this._run()), this.running.catch(this.keet.onerror)),
this._promise
)
}
runBackground() {
this.run().catch(this.keet.onerror)
}
}
function f(e) {
const t = { type: e.type, discoveryKey: e.discoveryKey }
return { gte: t, lte: t }
}
module.exports = class extends e {
constructor(e) {
;(super(),
(this.keet = e),
(this.local = this.keet.local),
(this._flushLock = new s({ debounce: !0 })),
(this._scopeLocks = new Map()),
(this._handles = []),
this.ready().catch(e.onerror))
}
async _open() {
if (!this.keet.readOnly)
for await (const e of this.local.find("@local/tasks")) {
const t = e.type,
s = e.discoveryKey
this._createHandle(t, s, e).runBackground()
}
}
_close() {
const e = []
for (const t of this._handles) e.push(t.close())
return Promise.allSettled(e)
}
async flush() {
await this.internalFlushTasks()
const e = []
for (const t of this._handles) e.push(t.running)
await Promise.allSettled(e)
}
async internalFlushTasks() {
if ((await this._flushLock.lock()) && !this.closing)
try {
const e = await this.keet.internalEnterLocalTransaction()
try {
let t = !1
for (const s of this._handles)
s.running &&
!s.completed &&
((await e.get("@local/tasks-by-type", s.description)) ||
(await e.insert("@local/tasks", s.description), (t = !0)))
t && (await e.flush())
} finally {
await this.keet.internalExitLocalTransaction()
}
} catch (e) {
this.keet.onerror(e)
} finally {
this._flushLock.unlock()
}
}
_createHandle(e, t, s, r = {}) {
const i = { ...s, type: e, discoveryKey: t, timestamp: Date.now(), priority: r.priority || g },
n = a.getId(i),
o = new R(this, n, i)
return ((o.idx = this._handles.push(o) - 1), o)
}
async internalHandleTaskStarted(e) {
let t = this._scopeLocks.get(e.scopeId)
;(t || ((t = new s()), this._scopeLocks.set(e.scopeId, t)), await t.lock())
}
async internalHandleTaskCompleted(e) {
if (this.closing) return
const t = this._handles.pop()
t !== e && ((t.idx = e.idx), (this._handles[e.idx] = t))
const s = await this.keet.internalEnterLocalTransaction()
try {
for await (const t of s.find("@local/tasks-by-type", f(e.description)))
await s.delete("@local/tasks", t)
await s.flush()
} catch (e) {
this.keet.onerror(e)
} finally {
await this.keet.internalExitLocalTransaction()
}
const r = this._scopeLocks.get(e.scopeId),
i = r.waiting.length
;(r.unlock(), 0 === i && this._scopeLocks.delete(e.scopeId))
}
restoreIdentity(e, t) {
return this._createHandle(o.RestoreIdentity, e, { restoreIdentity: t })
}
createRoom(e, t) {
return this._createHandle(o.CreateRoom, e, { createRoom: t })
}
leaveRoom(e, t) {
return this._createHandle(o.LeaveRoom, e, { leaveRoom: t })
}
clearRoom(e, t) {
return this._createHandle(o.ClearRoom, e, { clearRoom: t })
}
joinRoomInvitation(e, t) {
return this._createHandle(o.JoinRoomInvitation, e, { joinRoomInvitation: t })
}
joinRoomKeys(e, t) {
return this._createHandle(o.JoinRoomKeys, e, { joinRoomKeys: t })
}
joinRoomResponse(e, t) {
return this._createHandle(o.JoinRoomResponse, e, { joinRoomResponse: t })
}
sendContactRequest(e, t) {
return this._createHandle(o.SendContactRequest, e, { sendContactRequest: t })
}
testFastTask(e) {
return this._createHandle(o.TestFastTask, e)
}
testSlowTask(e) {
return this._createHandle(o.TestSlowTask, e)
}
testErrorTask(e) {
return this._createHandle(o.TestErrorTask, e)
}
}