fix memory leaks using typedarrays for cidr and streaming video uploads

src/utils/ranges.ts

@@ -51,17 +51,36 @@ interface AsnMatch {
   range: string | null
 }
 
-interface NetworkAsnRecord {
+class NetworksDb {
-  asn: number
+  ipv4Network: Uint32Array
-  cidr: string
+  ipv4Prefix: Uint8Array
-  parsed: ParsedCidr
+  ipv4Asn: Uint32Array
+  ipv4Count: number
+
+  ipv6Network: bigint[]
+  ipv6Prefix: Uint8Array
+  ipv6Asn: Uint32Array
+  ipv6Count: number
+
+  constructor(maxSize: number) {
+    this.ipv4Network = new Uint32Array(maxSize)
+    this.ipv4Prefix = new Uint8Array(maxSize)
+    this.ipv4Asn = new Uint32Array(maxSize)
+    this.ipv4Count = 0
+
+    this.ipv6Network = new Array(maxSize)
+    this.ipv6Prefix = new Uint8Array(maxSize)
+    this.ipv6Asn = new Uint32Array(maxSize)
+    this.ipv6Count = 0
+  }
 }
 
-let networksCache: NetworkAsnRecord[] | null = null
+let networksDb: NetworksDb | null = null
 let networkRefreshPromise: Promise<void> | null = null
 let networksCheckedAt = 0
 let networksEtag: string | null = null
 
+
 interface BlockedListCache {
   mtimeMs: number;
   parsedCidrs: { cidr: string, parsed: ParsedCidr }[];
@@ -178,7 +197,7 @@ function parseCidr(cidr: string): ParsedCidr | null {
   if (ip.version === 4) {
     if (prefix < 0 || prefix > 32) return null
     const mask = prefix === 0 ? 0 : (0xffffffff << (32 - prefix)) >>> 0
-    return { version: 4, network: ip.value & mask, mask, prefix }
+    return { version: 4, network: (ip.value & mask) >>> 0, mask, prefix }
   }
 
   if (prefix < 0 || prefix > 128) return null
@@ -195,7 +214,7 @@ function isIpInCidr(ip: ParsedIp, cidr: string): boolean {
   if (!parsed) return false
 
   if (ip.version === 4 && parsed.version === 4) {
-    return (ip.value & parsed.mask) === parsed.network
+    return ((ip.value & parsed.mask) >>> 0) === parsed.network
   }
 
   if (ip.version === 6 && parsed.version === 6) {
@@ -223,8 +242,13 @@ function extractCidrs(text: string): string[] {
   return cidrs
 }
 
-function parseCsvNetworks(text: string): NetworkAsnRecord[] {
+function parseCsvNetworks(text: string): NetworksDb {
-  const records: NetworkAsnRecord[] = []
+  let lineCount = 1
+  for (let i = 0; i < text.length; i++) {
+    if (text[i] === '\n') lineCount++
+  }
+
+  const db = new NetworksDb(lineCount)
   const lines = text.split(/\r?\n/)
 
   for (const line of lines) {
@@ -241,14 +265,21 @@ function parseCsvNetworks(text: string): NetworkAsnRecord[] {
     const parsed = parseCidr(cidr)
     if (!parsed) continue
 
-    records.push({
+    const asn = Number(asnText)
-      asn: Number(asnText),
+    if (parsed.version === 4) {
-      cidr,
+      const idx = db.ipv4Count++
-      parsed
+      db.ipv4Network[idx] = parsed.network
-    })
+      db.ipv4Prefix[idx] = parsed.prefix
+      db.ipv4Asn[idx] = asn
+    } else {
+      const idx = db.ipv6Count++
+      db.ipv6Network[idx] = parsed.network
+      db.ipv6Prefix[idx] = parsed.prefix
+      db.ipv6Asn[idx] = asn
+    }
   }
 
-  return records
+  return db
 }
 
 async function refreshNetworksCache(force: boolean = false): Promise<void> {
@@ -256,7 +287,7 @@ async function refreshNetworksCache(force: boolean = false): Promise<void> {
 
   networkRefreshPromise = (async () => {
     const isFresh =
-      networksCache !== null &&
+      networksDb !== null &&
       networksCheckedAt > 0 &&
       Date.now() - networksCheckedAt < NETWORKS_REFRESH_INTERVAL_MS
 
@@ -275,7 +306,7 @@ async function refreshNetworksCache(force: boolean = false): Promise<void> {
 
       const remoteEtag = headResponse.headers.get('etag')
       const shouldDownload =
-        networksCache === null || !remoteEtag || !networksEtag || remoteEtag !== networksEtag
+        networksDb === null || !remoteEtag || !networksEtag || remoteEtag !== networksEtag
 
       if (shouldDownload) {
         const downloadResponse = await fetch('https://ip.guide/bulk/networks.csv', {
@@ -284,15 +315,14 @@ async function refreshNetworksCache(force: boolean = false): Promise<void> {
         if (!downloadResponse.ok) {
           throw new Error(`failed to fetch ip.guide with ${downloadResponse.status}`)
         }
-
         const content = await downloadResponse.text()
-        networksCache = parseCsvNetworks(content)
+        networksDb = parseCsvNetworks(content)
       }
 
       networksCheckedAt = Date.now()
       networksEtag = remoteEtag ?? networksEtag
     } catch (error) {
-      if (networksCache === null) networksCache = []
+      if (networksDb === null) networksDb = new NetworksDb(0)
 
       console.error('Failed to refresh ASN network ranges', error)
     }
@@ -303,40 +333,62 @@ async function refreshNetworksCache(force: boolean = false): Promise<void> {
   return networkRefreshPromise
 }
 
+function intToIpv4(ip: number): string {
+  return `${(ip >>> 24) & 255}.${(ip >>> 16) & 255}.${(ip >>> 8) & 255}.${ip & 255}`
+}
+
+function bigintToIpv6(ip: bigint): string {
+  const parts: string[] = []
+  for (let i = 7n; i >= 0n; i--) {
+    parts.push(((ip >> (i * 16n)) & 0xffffn).toString(16))
+  }
+  return parts.join(':')
+}
+
 async function resolveIpAsn(parsedIp: ParsedIp): Promise<AsnMatch> {
   await refreshNetworksCache()
-  const records = networksCache ?? []
+  const db = networksDb
-  let bestMatch: NetworkAsnRecord | null = null
 
-  for (const record of records) {
+  let bestAsn: number | null = null
-    if (parsedIp.version === 4 && record.parsed.version === 4) {
+  let bestPrefix = -1
-      if ((parsedIp.value & record.parsed.mask) === record.parsed.network) {
+  let bestNetwork: number | bigint | null = null
-        if (bestMatch == null || record.parsed.prefix > bestMatch.parsed.prefix) {
-          bestMatch = record
-        }
-      }
-    }
 
-    if (parsedIp.version === 6 && record.parsed.version === 6) {
+  if (db && parsedIp.version === 4) {
-      if ((parsedIp.value & record.parsed.mask) === record.parsed.network) {
+    const value = parsedIp.value
-        if (bestMatch == null || record.parsed.prefix > bestMatch.parsed.prefix) {
+    for (let i = 0; i < db.ipv4Count; i++) {
-          bestMatch = record
+      const prefix = db.ipv4Prefix[i]
+      if (prefix > bestPrefix) {
+        const mask = prefix === 0 ? 0 : (0xffffffff << (32 - prefix)) >>> 0
+        if (((value & mask) >>> 0) === db.ipv4Network[i]) {
+          bestPrefix = prefix
+          bestAsn = db.ipv4Asn[i]
+          bestNetwork = db.ipv4Network[i]
+        }
+      }
+    }
+  } else if (db && parsedIp.version === 6) {
+    const value = parsedIp.value
+    for (let i = 0; i < db.ipv6Count; i++) {
+      const prefix = db.ipv6Prefix[i]
+      if (prefix > bestPrefix) {
+        const mask = prefix === 0 ? 0n : (IPV6_FULL_MASK ^ ((1n << (IPV6_BITS - BigInt(prefix))) - 1n)) & IPV6_FULL_MASK
+        if ((value & mask) === db.ipv6Network[i]) {
+          bestPrefix = prefix
+          bestAsn = db.ipv6Asn[i]
+          bestNetwork = db.ipv6Network[i]
         }
       }
     }
   }
 
-  if (bestMatch) {
+  if (bestAsn !== null) {
-    return {
+    const range = parsedIp.version === 4
-      asn: bestMatch.asn,
+      ? `${intToIpv4(bestNetwork as number)}/${bestPrefix}`
-      range: bestMatch.cidr
+      : `${bigintToIpv6(bestNetwork as bigint)}/${bestPrefix}`
-    }
+    return { asn: bestAsn, range }
   }
 
-  return {
+  return { asn: null, range: null }
-    asn: null,
-    range: null
-  }
 }
 
 export async function checkIpRanges(ip: string): Promise<BlockedIpResult> {
@@ -362,7 +414,7 @@ export async function checkIpRanges(ip: string): Promise<BlockedIpResult> {
       let matched = false;
       
       if (parsedIp.version === 4 && parsed.version === 4) {
-        matched = (parsedIp.value & parsed.mask) === parsed.network
+        matched = ((parsedIp.value & parsed.mask) >>> 0) === parsed.network
       } else if (parsedIp.version === 6 && parsed.version === 6) {
         matched = (parsedIp.value & parsed.mask) === parsed.network
       }

src/utils/redis.ts

@@ -16,7 +16,7 @@ setInterval(async () => {
   const targetFiles = files.filter((file) => file.endsWith('.webm') || file.endsWith('.mp4') || file.endsWith('.m4a'))
   targetFiles.forEach(async (f) => {
     const videoId = f.includes('_') ? f.split('_')[0] : f.replace('.mp4', '')
-    const isActive = await redis.get(videoId)
+    const isActive = await redis.get(`save:${videoId}`)
     if (!isActive) {
       fs.unlinkSync(`./videos/${f}`)
       console.log(`deleted file ${f} because there is no active download of it`)

src/utils/upload.ts

@@ -2,9 +2,12 @@ import * as fs from 'node:fs'
 const keys = JSON.parse(fs.readFileSync('s3.json', 'utf-8'))
 
 async function uploadVideo(video: string) {
-  const fileBuffer = await Bun.file(video).arrayBuffer()
+  const file = Bun.file(video)
+  const stream = fs.createReadStream(video)
   const hasher = new Bun.CryptoHasher("sha256");
-  hasher.update(fileBuffer);
+  for await (const chunk of stream) {
+    hasher.update(chunk);
+  }
   const fileHash = hasher.digest("hex");
 
   const uploaded = await fetch(`${keys.endpoint}/preservetube/${video.split('/')[2]}`, {
@@ -13,7 +16,7 @@ async function uploadVideo(video: string) {
       'x-authtoken': keys.videos[0].secret,
       'x-file-hash': fileHash
     },
-    body: fileBuffer
+    body: file
   })
   if (!uploaded.ok) throw new Error(`failed to upload video - ${uploaded.status} (${uploaded.statusText}) ${await uploaded.text()}`)
   return uploaded.url.replace(keys.endpoint, 'https://s5.archive.party')