// node examples/subscribe-position-openings.mjs https://dashboard.example ADDRESS_A,ADDRESS_B [all|raydium|orca]
// Requires the project's `ws` dependency. This example prints signals; it does not place orders.
import { existsSync, readFileSync, writeFileSync, renameSync } from 'node:fs'
import { WebSocket } from 'ws'

const [, , origin, creatorList, dex = 'all'] = process.argv
if (!origin || !creatorList) throw new Error('Usage: node examples/subscribe-position-openings.mjs DASHBOARD_ORIGIN ADDRESS_A[,ADDRESS_B,...] [all|raydium|orca]')
const requestedCreators = creatorList.split(',').map(address => address.trim())
if (requestedCreators.length > 100 || requestedCreators.some(address => !address)) throw new Error('Provide 1–100 non-empty addresses')
const creators = [...new Set(requestedCreators)].sort()
const base = new URL('/api/public/position-openings/ws', origin)
base.protocol = base.protocol === 'https:' || base.protocol === 'wss:' ? 'wss:' : 'ws:'
for (const creator of creators) base.searchParams.append('creator', creator)
base.searchParams.set('dex', dex)
const scope = base.toString()
const file = process.env.CURSOR_FILE || './position-opening-cursor.json'
const saved = existsSync(file) ? JSON.parse(readFileSync(file, 'utf8')) : null
if (saved && saved.scope !== scope) throw new Error('Use a different CURSOR_FILE for each origin/address-set/DEX subscription')
let cursor = saved?.cursor // omitted on first connection: live events only
let retryMs = 1000
let stopped = false
let current

async function handleOpening(event) {
  // Replace with your own durable, eventId-idempotent signal handler.
  // A replay is historical: check fundedAt freshness and source position state.
  // Use BigInt(event.tokenA.amountRaw) / BigInt(event.tokenB.amountRaw), chain ticks,
  // and a fresh DEX SDK quote to construct your own transaction and slippage limits.
  console.log(JSON.stringify(event))
}
async function handleClose(event) {
  // Resolve your own copied position through event.openingEventId (or source PDA).
  // Never close a different position just because its creator/pool matches.
  // Commit handling idempotently by event.eventId before advancing the cursor.
  console.log(JSON.stringify(event))
}
async function handleLiquidityChange(event) {
  // Match your copied position using openingEventId; deduplicate with eventId.
  // liquidityDelta and token amounts are unsigned magnitudes, not balances.
  // Requote with current DEX state; a decrease does not imply account closure.
  console.log(JSON.stringify(event))
}
function checkpoint(next) {
  writeFileSync(file + '.tmp', JSON.stringify({ scope, cursor: next }))
  renameSync(file + '.tmp', file)
  cursor = next
}
function connect() {
  if (stopped) return
  const url = new URL(base)
  if (cursor !== undefined) url.searchParams.set('after', cursor)
  const ws = current = new WebSocket(url)
  let work = Promise.resolve()
  let failed = false
  ws.on('open', () => { retryMs = 1000 })
  ws.on('message', raw => {
    work = work.then(async () => {
      if (failed || stopped) return
      const message = JSON.parse(raw.toString())
      if (message.type === 'position.opened') await handleOpening(message.data)
      if (message.type === 'position.closed') await handleClose(message.data)
      if (['position.increased', 'position.decreased'].includes(message.type)) await handleLiquidityChange(message.data)
      if (['ready', 'position.opened', 'position.closed', 'position.increased', 'position.decreased', 'checkpoint'].includes(message.type)) checkpoint(message.cursor)
    }).catch(error => { failed = true; console.error(error); ws.terminate() })
  })
  ws.on('unexpected-response', (_, response) => {
    console.error(`WebSocket handshake HTTP ${response.statusCode}`)
    // Invalid scope or missing history needs an explicit choice of a new starting point.
    if ([400, 409, 410].includes(response.statusCode)) stopped = true
    response.resume(); ws.terminate()
  })
  ws.on('error', error => console.error(error.message))
  ws.on('close', async code => {
    await work // never advance a new connection ahead of unfinished old messages
    if (code === 1008) { stopped = true; console.error('Subscription cursor/policy reset required') }
    if (stopped) return
    const wait = retryMs + Math.random() * 500
    retryMs = Math.min(30000, retryMs * 2)
    setTimeout(connect, wait)
  })
}
for (const signal of ['SIGINT', 'SIGTERM']) process.once(signal, () => { stopped = true; current?.close() })
connect()