Recipes & References

// Copy-paste the topic hash you need
const UNISWAP_V2_SWAP = "0xd78ad95fa46c994b6551d0da85fc275fe613ce37657fb8d5e3d130840159d822"

const logs = yield* getLogs({
  topics: [[UNISWAP_V2_SWAP]],
  fromBlock: "0x1000000",
  toBlock: "latest"
})

const RPC_URLS = [
  process.env.ALCHEMY_URL,           // Primary (if set)
  "https://eth.llamarpc.com",        // Fallback 1
  "https://rpc.ankr.com/eth",        // Fallback 2
  "https://ethereum.publicnode.com", // Fallback 3
].filter(Boolean) as string[]

// Topics are 32 bytes, addresses are 20 bytes (right-aligned)
const address = "0x" + topic.slice(-40)

// Each param is 32 bytes = 64 hex chars
const data = log.data.slice(2) // remove 0x
const param0 = BigInt("0x" + data.slice(0, 64))
const param1 = BigInt("0x" + data.slice(64, 128))
const param2 = BigInt("0x" + data.slice(128, 192))

const raw = BigInt("0x" + data.slice(0, 64))
// Check sign bit (most significant bit)
const int256 = raw >= 2n ** 255n ? raw - 2n ** 256n : raw

const bytes32 = "0x" + data.slice(0, 64)

// First 32 bytes = offset to data
const offset = parseInt(data.slice(0, 64), 16) * 2
// Next 32 bytes at offset = length
const length = parseInt(data.slice(offset, offset + 64), 16) * 2
// Actual data follows
const stringHex = data.slice(offset + 64, offset + 64 + length)
const string = Buffer.from(stringHex, 'hex').toString('utf8')

import { Schedule } from "effect"

const retrySchedule = Schedule.exponential("100 millis").pipe(
  Schedule.jittered,
  Schedule.compose(Schedule.recurs(3))
)

myEffect.pipe(Effect.retry(retrySchedule))

myEffect.pipe(Effect.timeout("5 seconds"))

const tryAll = <A, E>(effects: Effect.Effect<A, E>[]) =>
  effects.slice(1).reduce(
    (acc, eff) => acc.pipe(Effect.orElse(() => eff)),
    effects[0]!
  )

Effect.forEach(items, processItem, { concurrency: 10 })

Stream.runCollect(myStream).pipe(Effect.map(Chunk.toArray))