1import fs from 'fs/promises'
2import { existsSync, createWriteStream } from 'fs'
3import path from 'path'
4import { Readable } from 'stream'
5import { pipeline } from 'stream/promises'
6import { ossEnsName } from '../../sharedF/ossRepoConfigF.mts'
7import { sleep } from '../../sharedF/sleepF.mts'
8
9export const mediaSiteBaseUrlA = [`https://${ossEnsName}.demokluster.com`, `https://${ossEnsName.replace('.eth', '')}.eth.limo`]
10
11export const isTransientStreamErr = (err: any): boolean => {
12  const code = err?.code ?? err?.cause?.code
13  if (code === 'ERR_HTTP2_STREAM_ERROR' || code === 'ECONNRESET' || code === 'ETIMEDOUT') return true
14  const msg = String(err?.message ?? '')
15  return msg.includes('terminated') || msg.includes('other side closed')
16}
17
18export const fetchMediaWithFallback = async (relPath: string, {fetchImpl = fetch, baseUrlA = mediaSiteBaseUrlA}: {fetchImpl?: typeof fetch, baseUrlA?: string[]} = {}) => {
19  for (const baseUrl of baseUrlA) {
20    for (let attempt = 0; attempt < 2; attempt++) {
21      try {
22        const resp = await fetchImpl(`${baseUrl}/${relPath}`)
23        if (resp.ok) return resp
24        await resp.text().catch(() => {})
25        break
26      } catch (err: any) {
27        if (attempt === 0 && isTransientStreamErr(err)) {
28          console.log(`[media-proxy] transient upstream error for ${relPath}: ${err?.message ?? err} — retrying`)
29          await sleep(1000)
30          continue
31        }
32        break
33      }
34    }
35  }
36  return null
37}
38
39const inflightH: Record<string, Promise<boolean>> = {}
40
41const fetchToDiskOnce = async ({relPath, destPath, baseUrlA, fetchImpl}: {relPath: string, destPath: string, baseUrlA: string[], fetchImpl: typeof fetch}): Promise<boolean> => {
42  const partial = `${destPath}.partial`
43  try {
44    const resp = await fetchMediaWithFallback(relPath, {fetchImpl, baseUrlA})
45    if (!resp) return false
46    await fs.mkdir(path.dirname(destPath), {recursive: true})
47    await pipeline(Readable.fromWeb(resp.body as any), createWriteStream(partial))
48    await fs.rename(partial, destPath)
49    return true
50  } catch (err) {
51    await fs.rm(partial, {force: true}).catch(() => {})
52    throw err
53  }
54}
55
56const fetchToDisk = async (opts: {relPath: string, destPath: string, baseUrlA: string[], fetchImpl: typeof fetch}): Promise<boolean> => {
57  const maxAttempts = 3
58  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
59    try {
60      return await fetchToDiskOnce(opts)
61    } catch (err: any) {
62      if (attempt === maxAttempts) return false
63      console.log(`[media-proxy] upstream stream error for ${opts.relPath} (attempt ${attempt}/${maxAttempts}): ${err?.message ?? err}`)
64      await sleep(1000)
65    }
66  }
67  return false
68}
69
70// Write-through disk cache for the devserver proxy: first request downloads to destPath
71// (concurrent requests dedup via inflightH), later requests serve the file statically.
72export const proxyMediaCached = async ({relPath, destPath, baseUrlA = mediaSiteBaseUrlA, fetchImpl = fetch}: {relPath: string, destPath: string, baseUrlA?: string[], fetchImpl?: typeof fetch}): Promise<boolean> => {
73  if (existsSync(destPath)) return true
74  inflightH[destPath] ??= fetchToDisk({relPath, destPath, baseUrlA, fetchImpl}).finally(() => { delete inflightH[destPath] })
75  return inflightH[destPath]
76}
77
78const mediaSitePrefixH: [string, string][] = [
79  ['threerulecycle/', 'pt0/threerulecycleapp/public/'],
80  ['blogposts/', 'pt0/peatsite/blogposts/'],
81]
82
83export const resolveMediaRepoPath = (siteRelPath: string): string => {
84  for (const [sitePrefix, repoPrefix] of mediaSitePrefixH) {
85    if (siteRelPath.startsWith(sitePrefix)) return repoPrefix + siteRelPath.slice(sitePrefix.length)
86  }
87  return `pt0/peatsite/public/${siteRelPath}`
88}
89