sync.tsx 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378
  1. import { batch, createMemo } from "solid-js"
  2. import { createStore, produce, reconcile } from "solid-js/store"
  3. import { Binary } from "@opencode-ai/util/binary"
  4. import { retry } from "@opencode-ai/util/retry"
  5. import { createSimpleContext } from "@opencode-ai/ui/context"
  6. import { useGlobalSync } from "./global-sync"
  7. import { useSDK } from "./sdk"
  8. import type { Message, Part } from "@opencode-ai/sdk/v2/client"
  9. function sortParts(parts: Part[]) {
  10. return parts.filter((part) => !!part?.id).sort((a, b) => cmp(a.id, b.id))
  11. }
  12. function runInflight(map: Map<string, Promise<void>>, key: string, task: () => Promise<void>) {
  13. const pending = map.get(key)
  14. if (pending) return pending
  15. const promise = task().finally(() => {
  16. map.delete(key)
  17. })
  18. map.set(key, promise)
  19. return promise
  20. }
  21. const keyFor = (directory: string, id: string) => `${directory}\n${id}`
  22. const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0)
  23. type OptimisticStore = {
  24. message: Record<string, Message[] | undefined>
  25. part: Record<string, Part[] | undefined>
  26. }
  27. type OptimisticAddInput = {
  28. sessionID: string
  29. message: Message
  30. parts: Part[]
  31. }
  32. type OptimisticRemoveInput = {
  33. sessionID: string
  34. messageID: string
  35. }
  36. export function applyOptimisticAdd(draft: OptimisticStore, input: OptimisticAddInput) {
  37. const messages = draft.message[input.sessionID]
  38. if (!messages) {
  39. draft.message[input.sessionID] = [input.message]
  40. }
  41. if (messages) {
  42. const result = Binary.search(messages, input.message.id, (m) => m.id)
  43. messages.splice(result.index, 0, input.message)
  44. }
  45. draft.part[input.message.id] = sortParts(input.parts)
  46. }
  47. export function applyOptimisticRemove(draft: OptimisticStore, input: OptimisticRemoveInput) {
  48. const messages = draft.message[input.sessionID]
  49. if (messages) {
  50. const result = Binary.search(messages, input.messageID, (m) => m.id)
  51. if (result.found) messages.splice(result.index, 1)
  52. }
  53. delete draft.part[input.messageID]
  54. }
  55. function setOptimisticAdd(setStore: (...args: unknown[]) => void, input: OptimisticAddInput) {
  56. setStore("message", input.sessionID, (messages: Message[] | undefined) => {
  57. if (!messages) return [input.message]
  58. const result = Binary.search(messages, input.message.id, (m) => m.id)
  59. const next = [...messages]
  60. next.splice(result.index, 0, input.message)
  61. return next
  62. })
  63. setStore("part", input.message.id, sortParts(input.parts))
  64. }
  65. function setOptimisticRemove(setStore: (...args: unknown[]) => void, input: OptimisticRemoveInput) {
  66. setStore("message", input.sessionID, (messages: Message[] | undefined) => {
  67. if (!messages) return messages
  68. const result = Binary.search(messages, input.messageID, (m) => m.id)
  69. if (!result.found) return messages
  70. const next = [...messages]
  71. next.splice(result.index, 1)
  72. return next
  73. })
  74. setStore("part", (part: Record<string, Part[] | undefined>) => {
  75. if (!(input.messageID in part)) return part
  76. const next = { ...part }
  77. delete next[input.messageID]
  78. return next
  79. })
  80. }
  81. export const { use: useSync, provider: SyncProvider } = createSimpleContext({
  82. name: "Sync",
  83. init: () => {
  84. const globalSync = useGlobalSync()
  85. const sdk = useSDK()
  86. type Child = ReturnType<(typeof globalSync)["child"]>
  87. type Setter = Child[1]
  88. const current = createMemo(() => globalSync.child(sdk.directory))
  89. const target = (directory?: string) => {
  90. if (!directory || directory === sdk.directory) return current()
  91. return globalSync.child(directory)
  92. }
  93. const absolute = (path: string) => (current()[0].path.directory + "/" + path).replace("//", "/")
  94. const messagePageSize = 400
  95. const inflight = new Map<string, Promise<void>>()
  96. const inflightDiff = new Map<string, Promise<void>>()
  97. const inflightTodo = new Map<string, Promise<void>>()
  98. const [meta, setMeta] = createStore({
  99. limit: {} as Record<string, number>,
  100. complete: {} as Record<string, boolean>,
  101. loading: {} as Record<string, boolean>,
  102. })
  103. const getSession = (sessionID: string) => {
  104. const store = current()[0]
  105. const match = Binary.search(store.session, sessionID, (s) => s.id)
  106. if (match.found) return store.session[match.index]
  107. return undefined
  108. }
  109. const limitFor = (count: number) => {
  110. if (count <= messagePageSize) return messagePageSize
  111. return Math.ceil(count / messagePageSize) * messagePageSize
  112. }
  113. const fetchMessages = async (input: { client: typeof sdk.client; sessionID: string; limit: number }) => {
  114. const messages = await retry(() =>
  115. input.client.session.messages({ sessionID: input.sessionID, limit: input.limit }),
  116. )
  117. const items = (messages.data ?? []).filter((x) => !!x?.info?.id)
  118. const session = items
  119. .map((x) => x.info)
  120. .filter((m) => !!m?.id)
  121. .sort((a, b) => cmp(a.id, b.id))
  122. const part = items.map((message) => ({ id: message.info.id, part: sortParts(message.parts) }))
  123. return {
  124. session,
  125. part,
  126. complete: session.length < input.limit,
  127. }
  128. }
  129. const loadMessages = async (input: {
  130. directory: string
  131. client: typeof sdk.client
  132. setStore: Setter
  133. sessionID: string
  134. limit: number
  135. }) => {
  136. const key = keyFor(input.directory, input.sessionID)
  137. if (meta.loading[key]) return
  138. setMeta("loading", key, true)
  139. await fetchMessages(input)
  140. .then((next) => {
  141. batch(() => {
  142. input.setStore("message", input.sessionID, reconcile(next.session, { key: "id" }))
  143. for (const message of next.part) {
  144. input.setStore("part", message.id, reconcile(message.part, { key: "id" }))
  145. }
  146. setMeta("limit", key, input.limit)
  147. setMeta("complete", key, next.complete)
  148. })
  149. })
  150. .finally(() => {
  151. setMeta("loading", key, false)
  152. })
  153. }
  154. return {
  155. get data() {
  156. return current()[0]
  157. },
  158. get set(): Setter {
  159. return current()[1]
  160. },
  161. get status() {
  162. return current()[0].status
  163. },
  164. get ready() {
  165. return current()[0].status !== "loading"
  166. },
  167. get project() {
  168. const store = current()[0]
  169. const match = Binary.search(globalSync.data.project, store.project, (p) => p.id)
  170. if (match.found) return globalSync.data.project[match.index]
  171. return undefined
  172. },
  173. session: {
  174. get: getSession,
  175. optimistic: {
  176. add(input: { directory?: string; sessionID: string; message: Message; parts: Part[] }) {
  177. const [, setStore] = target(input.directory)
  178. setOptimisticAdd(setStore as (...args: unknown[]) => void, input)
  179. },
  180. remove(input: { directory?: string; sessionID: string; messageID: string }) {
  181. const [, setStore] = target(input.directory)
  182. setOptimisticRemove(setStore as (...args: unknown[]) => void, input)
  183. },
  184. },
  185. addOptimisticMessage(input: {
  186. sessionID: string
  187. messageID: string
  188. parts: Part[]
  189. agent: string
  190. model: { providerID: string; modelID: string }
  191. }) {
  192. const message: Message = {
  193. id: input.messageID,
  194. sessionID: input.sessionID,
  195. role: "user",
  196. time: { created: Date.now() },
  197. agent: input.agent,
  198. model: input.model,
  199. }
  200. const [, setStore] = target()
  201. setOptimisticAdd(setStore as (...args: unknown[]) => void, {
  202. sessionID: input.sessionID,
  203. message,
  204. parts: input.parts,
  205. })
  206. },
  207. async sync(sessionID: string) {
  208. const directory = sdk.directory
  209. const client = sdk.client
  210. const [store, setStore] = globalSync.child(directory)
  211. const key = keyFor(directory, sessionID)
  212. const hasSession = (() => {
  213. const match = Binary.search(store.session, sessionID, (s) => s.id)
  214. return match.found
  215. })()
  216. const hasMessages = store.message[sessionID] !== undefined
  217. const hydrated = meta.limit[key] !== undefined
  218. if (hasSession && hasMessages && hydrated) return
  219. const count = store.message[sessionID]?.length ?? 0
  220. const limit = hydrated ? (meta.limit[key] ?? messagePageSize) : limitFor(count)
  221. const sessionReq = hasSession
  222. ? Promise.resolve()
  223. : retry(() => client.session.get({ sessionID })).then((session) => {
  224. const data = session.data
  225. if (!data) return
  226. setStore(
  227. "session",
  228. produce((draft) => {
  229. const match = Binary.search(draft, sessionID, (s) => s.id)
  230. if (match.found) {
  231. draft[match.index] = data
  232. return
  233. }
  234. draft.splice(match.index, 0, data)
  235. }),
  236. )
  237. })
  238. const messagesReq =
  239. hasMessages && hydrated
  240. ? Promise.resolve()
  241. : loadMessages({
  242. directory,
  243. client,
  244. setStore,
  245. sessionID,
  246. limit,
  247. })
  248. return runInflight(inflight, key, () => Promise.all([sessionReq, messagesReq]).then(() => {}))
  249. },
  250. async diff(sessionID: string) {
  251. const directory = sdk.directory
  252. const client = sdk.client
  253. const [store, setStore] = globalSync.child(directory)
  254. if (store.session_diff[sessionID] !== undefined) return
  255. const key = keyFor(directory, sessionID)
  256. return runInflight(inflightDiff, key, () =>
  257. retry(() => client.session.diff({ sessionID })).then((diff) => {
  258. setStore("session_diff", sessionID, reconcile(diff.data ?? [], { key: "file" }))
  259. }),
  260. )
  261. },
  262. async todo(sessionID: string) {
  263. const directory = sdk.directory
  264. const client = sdk.client
  265. const [store, setStore] = globalSync.child(directory)
  266. const existing = store.todo[sessionID]
  267. if (existing !== undefined) {
  268. if (globalSync.data.session_todo[sessionID] === undefined) {
  269. globalSync.todo.set(sessionID, existing)
  270. }
  271. return
  272. }
  273. const cached = globalSync.data.session_todo[sessionID]
  274. if (cached !== undefined) {
  275. setStore("todo", sessionID, reconcile(cached, { key: "id" }))
  276. }
  277. const key = keyFor(directory, sessionID)
  278. return runInflight(inflightTodo, key, () =>
  279. retry(() => client.session.todo({ sessionID })).then((todo) => {
  280. const list = todo.data ?? []
  281. setStore("todo", sessionID, reconcile(list, { key: "id" }))
  282. globalSync.todo.set(sessionID, list)
  283. }),
  284. )
  285. },
  286. history: {
  287. more(sessionID: string) {
  288. const store = current()[0]
  289. const key = keyFor(sdk.directory, sessionID)
  290. if (store.message[sessionID] === undefined) return false
  291. if (meta.limit[key] === undefined) return false
  292. if (meta.complete[key]) return false
  293. return true
  294. },
  295. loading(sessionID: string) {
  296. const key = keyFor(sdk.directory, sessionID)
  297. return meta.loading[key] ?? false
  298. },
  299. async loadMore(sessionID: string, count = messagePageSize) {
  300. const directory = sdk.directory
  301. const client = sdk.client
  302. const [, setStore] = globalSync.child(directory)
  303. const key = keyFor(directory, sessionID)
  304. if (meta.loading[key]) return
  305. if (meta.complete[key]) return
  306. const currentLimit = meta.limit[key] ?? messagePageSize
  307. await loadMessages({
  308. directory,
  309. client,
  310. setStore,
  311. sessionID,
  312. limit: currentLimit + count,
  313. })
  314. },
  315. },
  316. fetch: async (count = 10) => {
  317. const directory = sdk.directory
  318. const client = sdk.client
  319. const [store, setStore] = globalSync.child(directory)
  320. setStore("limit", (x) => x + count)
  321. await client.session.list().then((x) => {
  322. const sessions = (x.data ?? [])
  323. .filter((s) => !!s?.id)
  324. .sort((a, b) => cmp(a.id, b.id))
  325. .slice(0, store.limit)
  326. setStore("session", reconcile(sessions, { key: "id" }))
  327. })
  328. },
  329. more: createMemo(() => current()[0].session.length >= current()[0].limit),
  330. archive: async (sessionID: string) => {
  331. const directory = sdk.directory
  332. const client = sdk.client
  333. const [, setStore] = globalSync.child(directory)
  334. await client.session.update({ sessionID, time: { archived: Date.now() } })
  335. setStore(
  336. produce((draft) => {
  337. const match = Binary.search(draft.session, sessionID, (s) => s.id)
  338. if (match.found) draft.session.splice(match.index, 1)
  339. }),
  340. )
  341. },
  342. },
  343. absolute,
  344. get directory() {
  345. return current()[0].path.directory
  346. },
  347. }
  348. },
  349. })