api.ts 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210
  1. import { DurableObject } from "cloudflare:workers"
  2. import { randomUUID } from "node:crypto"
  3. type Env = {
  4. SYNC_SERVER: DurableObjectNamespace<SyncServer>
  5. Bucket: R2Bucket
  6. }
  7. export class SyncServer extends DurableObject<Env> {
  8. constructor(ctx: DurableObjectState, env: Env) {
  9. super(ctx, env)
  10. }
  11. async fetch() {
  12. console.log("SyncServer subscribe")
  13. const webSocketPair = new WebSocketPair()
  14. const [client, server] = Object.values(webSocketPair)
  15. this.ctx.acceptWebSocket(server)
  16. const data = await this.ctx.storage.list()
  17. for (const [key, content] of data.entries()) {
  18. server.send(JSON.stringify({ key, content }))
  19. }
  20. return new Response(null, {
  21. status: 101,
  22. webSocket: client,
  23. })
  24. }
  25. async webSocketMessage(ws, message) {}
  26. async webSocketClose(ws, code, reason, wasClean) {
  27. ws.close(code, "Durable Object is closing WebSocket")
  28. }
  29. async publish(secret: string, key: string, content: any) {
  30. if (secret !== (await this.getSecret())) throw new Error("Invalid secret")
  31. const sessionID = await this.getSessionID()
  32. if (
  33. !key.startsWith(`session/info/${sessionID}`) &&
  34. !key.startsWith(`session/message/${sessionID}/`)
  35. )
  36. return new Response("Error: Invalid key", { status: 400 })
  37. // store message
  38. await this.env.Bucket.put(`share/${key}.json`, JSON.stringify(content), {
  39. httpMetadata: {
  40. contentType: "application/json",
  41. },
  42. })
  43. await this.ctx.storage.put(key, content)
  44. const clients = this.ctx.getWebSockets()
  45. console.log("SyncServer publish", key, "to", clients.length, "subscribers")
  46. for (const client of clients) {
  47. client.send(JSON.stringify({ key, content }))
  48. }
  49. }
  50. public async share(sessionID: string) {
  51. let secret = await this.getSecret()
  52. if (secret) return secret
  53. secret = randomUUID()
  54. await this.ctx.storage.put("secret", secret)
  55. await this.ctx.storage.put("sessionID", sessionID)
  56. return secret
  57. }
  58. public async getData() {
  59. const data = await this.ctx.storage.list()
  60. const messages = []
  61. for (const [key, content] of data.entries()) {
  62. messages.push({ key, content })
  63. }
  64. return messages
  65. }
  66. private async getSecret() {
  67. return this.ctx.storage.get<string>("secret")
  68. }
  69. private async getSessionID() {
  70. return this.ctx.storage.get<string>("sessionID")
  71. }
  72. async clear(secret: string) {
  73. await this.assertSecret(secret)
  74. await this.ctx.storage.deleteAll()
  75. }
  76. private async assertSecret(secret: string) {
  77. if (secret !== (await this.getSecret())) throw new Error("Invalid secret")
  78. }
  79. static shortName(id: string) {
  80. return id.substring(id.length - 8)
  81. }
  82. }
  83. export default {
  84. async fetch(request: Request, env: Env, ctx: ExecutionContext) {
  85. const url = new URL(request.url)
  86. const splits = url.pathname.split("/")
  87. const method = splits[1]
  88. if (request.method === "GET" && method === "") {
  89. return new Response("Hello, world!", {
  90. headers: { "Content-Type": "text/plain" },
  91. })
  92. }
  93. if (request.method === "POST" && method === "share_create") {
  94. const body = await request.json<any>()
  95. const sessionID = body.sessionID
  96. const short = SyncServer.shortName(sessionID)
  97. const id = env.SYNC_SERVER.idFromName(short)
  98. const stub = env.SYNC_SERVER.get(id)
  99. const secret = await stub.share(sessionID)
  100. return new Response(
  101. JSON.stringify({
  102. secret,
  103. url: "https://dev.opencode.ai/s/" + short,
  104. }),
  105. {
  106. headers: { "Content-Type": "application/json" },
  107. },
  108. )
  109. }
  110. if (request.method === "POST" && method === "share_delete") {
  111. const body = await request.json<any>()
  112. const sessionID = body.sessionID
  113. const secret = body.secret
  114. const id = env.SYNC_SERVER.idFromName(SyncServer.shortName(sessionID))
  115. const stub = env.SYNC_SERVER.get(id)
  116. await stub.clear(secret)
  117. return new Response(JSON.stringify({}), {
  118. headers: { "Content-Type": "application/json" },
  119. })
  120. }
  121. if (request.method === "POST" && method === "share_sync") {
  122. const body = await request.json<{
  123. sessionID: string
  124. secret: string
  125. key: string
  126. content: any
  127. }>()
  128. const name = SyncServer.shortName(body.sessionID)
  129. const id = env.SYNC_SERVER.idFromName(name)
  130. const stub = env.SYNC_SERVER.get(id)
  131. await stub.publish(body.secret, body.key, body.content)
  132. return new Response(JSON.stringify({}), {
  133. headers: { "Content-Type": "application/json" },
  134. })
  135. }
  136. if (request.method === "GET" && method === "share_poll") {
  137. const upgradeHeader = request.headers.get("Upgrade")
  138. if (!upgradeHeader || upgradeHeader !== "websocket") {
  139. return new Response("Error: Upgrade header is required", {
  140. status: 426,
  141. })
  142. }
  143. const id = url.searchParams.get("id")
  144. console.log("share_poll", id)
  145. if (!id)
  146. return new Response("Error: Share ID is required", { status: 400 })
  147. const stub = env.SYNC_SERVER.get(env.SYNC_SERVER.idFromName(id))
  148. return stub.fetch(request)
  149. }
  150. if (request.method === "GET" && method === "share_data") {
  151. const id = url.searchParams.get("id")
  152. console.log("share_data", id)
  153. if (!id)
  154. return new Response("Error: Share ID is required", { status: 400 })
  155. const stub = env.SYNC_SERVER.get(env.SYNC_SERVER.idFromName(id))
  156. const data = await stub.getData()
  157. let info
  158. const messages: Record<string, any> = {}
  159. data.forEach((d) => {
  160. const [root, type, ...splits] = d.key.split("/")
  161. if (root !== "session") return
  162. if (type === "info") {
  163. info = d.content
  164. return
  165. }
  166. if (type === "message") {
  167. const [, messageID] = splits
  168. messages[messageID] = d.content
  169. }
  170. })
  171. return new Response(
  172. JSON.stringify({
  173. info,
  174. messages,
  175. }),
  176. {
  177. headers: { "Content-Type": "application/json" },
  178. },
  179. )
  180. }
  181. },
  182. }