select *
  from read_json('activity/analytics/events-2024-00000-of-00001.json',
    format = 'newline_delimited',
    auto_detect = true)
  where (json->'predicted_gender') is not null;

select *
  from read_json('activity/analytics/events-2024-00000-of-00001.json',
    format = 'newline_delimited',
    columns = {
      predicted_gender: 'text',
      prob_male: 'double',
      prob_female: 'double',
      prob_non_binary_gender_expansive: 'double',
      day_pt: 'timestamp'})
  where predicted_gender is not null;

┌──────────────────┬────────────────────┬──────────────────────┬──────────────────────────────────┬─────────────────────┐
│ predicted_gender │     prob_male      │     prob_female      │ prob_non_binary_gender_expansive │       day_pt        │
│     varchar      │       double       │        double        │              double              │      timestamp      │
├──────────────────┼────────────────────┼──────────────────────┼──────────────────────────────────┼─────────────────────┤
│ male             │ 0.8108658790588379 │  0.11857519298791885 │              0.07055895030498505 │ 2023-10-18 00:00:00 │
│ male             │ 0.7923107147216797 │   0.1035185381770134 │              0.10417073965072632 │ 2023-12-13 00:00:00 │
│ male             │ 0.6698217391967773 │  0.14042726159095764 │              0.18975096940994263 │ 2024-02-07 00:00:00 │
│ male             │  0.734815776348114 │  0.08487804979085922 │               0.1803061068058014 │ 2024-04-03 00:00:00 │
│ male             │ 0.7945181131362915 │  0.16594618558883667 │             0.039535682648420334 │ 2023-01-11 00:00:00 │
│ male             │ 0.7358677387237549 │   0.1554252803325653 │              0.10870702564716339 │ 2024-05-29 00:00:00 │
                                                           [...]
│ male             │ 0.8582155108451843 │  0.07383262366056442 │              0.06795187294483185 │ 2023-11-08 00:00:00 │
│ male             │ 0.6448849439620972 │  0.19893291592597961 │               0.1561821550130844 │ 2024-01-03 00:00:00 │
│ male             │  0.720182478427887 │  0.09869085997343063 │               0.1811266392469406 │ 2024-02-28 00:00:00 │
├──────────────────┴────────────────────┴──────────────────────┴──────────────────────────────────┴─────────────────────┤
│ 78 rows (40 shown)                                                                                          5 columns │
└───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘

78 rows (40 shown)

select count(*)
  from read_json('activity/analytics/events-2024-00000-of-00001.json',
    format = 'newline_delimited',
    auto_detect = true)
  where (json->'predicted_gender') is not null or (json->'predicted_age') is not null;

166

let blob = await openAsBlob(process.argv[2])

// Anything past the final newline in a string chunk is part of an incomplete line and
// will get saved here.
let leftover = ''
let eventsFound = 0

let rd = blob.stream().pipeThrough(textDecoderStream()).getReader()
while (true) {
  let { chunk, done } = await rd.read()
  if (chunk === undefined) {
    break
  }

  if (leftover.length !== 0) {
    chunk = leftover + chunk
  }

  let offset = 0
  while (true) {
    let candidatePos = chunk.indexOf('model_version', offset)
    if (candidatePos === -1) {
      break
    }

    let lineEnd = chunk.indexOf('\n', candidatePos)
    if (lineEnd === -1) {
      break
    }

    let lineStart = chunk.lastIndexOf('\n', candidatePos)
    lineStart = lineStart !== -1 ? lineStart + 1 : 0

    let line = chunk.substring(lineStart, lineEnd)
    let event = JSON.parse(line)
    console.log(event)

    eventsFound++
    offset = lineEnd + 1
  }

  let nl = chunk.lastIndexOf('\n')
  if (nl === -1) {
    leftover = chunk
  } else {
    leftover = chunk.substring(nl + 1)
  }

  if (done) {
    break
  }
}

// Nothing should be left over past the final newline.
assert(leftover === '')

console.log({ eventsFound })

$ hyperfine {node,bun}' loader.mjs events-2024-00000-of-00001.json'
Benchmark 1: node loader.mjs events-2024-00000-of-00001.json
  Time (mean ± σ):      6.793 s ±  0.091 s    [User: 5.426 s, System: 8.493 s]
  Range (min … max):    6.647 s …  6.960 s    10 runs

Benchmark 2: bun loader.mjs events-2024-00000-of-00001.json
  Time (mean ± σ):      2.182 s ±  0.023 s    [User: 2.469 s, System: 0.437 s]
  Range (min … max):    2.152 s …  2.221 s    10 runs

#[wasm_bindgen]
struct A {
  pub x: i32,
  pub y: i32,
}

let a = somethingThatReturnsA()
try {
  // Use a.x and a.y
} finally {
  a.free()
}

#[wasm_bindgen(getter_with_clone)]
struct B {
  pub inner_a: A,
}

let b = somethingThatReturnsB()
try {
  let a = b.inner_a  // clones A
  try {
    // ...
  } finally {
    a.free()
  }
} finally {
  a.free()
}

#[wasm_bindgen(getter_with_clone)]
struct B {
  #[wasm_bindgen(js_name = innerA)]
  pub inner_a: A,
}

// I don't really get how package names work exactly, so I just went with "local" as
// the namespace like some examples used.
package local:discordsona;

interface loader {
  // Records are POD value types.
  record gender-predictions {
    tally-female: u32,
    tally-male: u32,
    tally-non-binary-gender-expansive: u32,

    day-pt: list<f64>,
    prob-female: list<f64>,
    prob-male: list<f64>,
    prob-non-binary-gender-expansive: list<f64>,
  }

  record age-predictions {
    tally-y13-y17: u32,
    tally-y18-y24: u32,
    tally-y25-y34: u32,
    tally-y35-over: u32,

    day-pt: list<f64>,
    prob-y13-y17: list<f64>,
    prob-y18-y24: list<f64>,
    prob-y25-y34: list<f64>,
    prob-y35-over: list<f64>,
  }

  record predictions {
    age: age-predictions,
    gender: gender-predictions,
  }

  record next-buffer {
    ptr: u32,
    size: u32,
  }

  // Resources are dynamically allocated objects with methods.
  resource loader {
    constructor(buffer-size: u32);
    process-buffer: func(size: u32) -> result<next-buffer, string>;
    // No built-in way of taking ownership from 'self' so a static method is needed. D:
    finish: static func(self: loader) -> predictions;
  }
}

// Idk what worlds are semantically, where is the name even used?
world w {
  export loader;
}

struct LoaderImpl { /* [...] */ }

impl LoaderImpl {
  fn new(buffer_size: u32) -> Self { /* [...] */ }
  fn process_buffer(&mut self, size: u32) -> Result<NextBuffer, String> { /* [...] */ }
  fn finish(self) -> Predictions { /* [...] */ }
}

struct CellWrapper<T>(RefCell<T>);

impl GuestLoader for CellWrapper<LoaderImpl> {
  fn new(buffer_size: u32) -> Self {
    Self(RefCell::new(LoaderImpl::new(buffer_size)))
  }

  fn finish(self_: Loader) -> Predictions {
    self_.into_inner::<Self>().0.into_inner().finish()
  }

  delegate! {
    to self.0.borrow_mut() {
      fn process_buffer(&self, size: u32) -> Result<NextBuffer, String>;
    }
  }
}

struct World;

impl Guest for World {
  type Loader = CellWrapper<LoaderImpl>;
}

struct LoaderImpl {
  buffer: Vec<u8>,
  leftover: usize,

  finder: memmem::Finder<'static>,
  json_buffers: simd_json::Buffers,
  predictions: Predictions,
}

impl LoaderImpl {
  fn new(buffer_size: u32) -> Self {
    Self {
      buffer: vec![0; buffer_size as usize],
      leftover: 0,
      finder: memmem::Finder::new("model_version"),
      // is buffer_size a good size for this?
      json_buffers: simd_json::Buffers::new(buffer_size as usize),
      predictions: Default::default(),
    }
  }
  /* [...] */
}

impl LoaderImpl {
  /* [...] */
  fn process_buffer(&mut self, size: u32) -> Result<NextBuffer, String> {
    let size = size as usize;
    if self.leftover + size > self.buffer.len() {
      return Err(format!(
        "{} + {} > {}",
        self.leftover,
        size,
        self.buffer.len()
      ));
    }

    let buffer = &mut self.buffer[..self.leftover + size];
    let mut remaining = &mut buffer[..];

    while !remaining.is_empty() {
      let Some(candidate) = self.finder.find(remaining) else {
        break;
      };

      let Some(line_end) =
        memchr(b'\n', &remaining[candidate..]).map(|n| n + candidate)
      else {
        break;
      };

      let line_start = memrchr(b'\n', &remaining[..candidate]).map_or(0, |i| i + 1);
      if let Ok(event) = simd_json::serde::from_slice_with_buffers::<Event>(
        &mut remaining[line_start..line_end],
        &mut self.json_buffers,
      ) {
        /* [...actually fill in self.predictions from the event...] */
      }

      remaining = &mut remaining[line_end + 1..];
    }

    if let Some(last_newline) = memrchr(b'\n', buffer) {
      let last_line_start = last_newline + 1;
      self.leftover = buffer.len() - last_line_start;
      if self.leftover > 0 {
        buffer.copy_within(last_line_start.., 0);
      }
    } else {
      if buffer.len() == self.buffer.len() {
        self.buffer.resize(self.buffer.len() * 2, 0);
      }

      self.leftover += size;
    }

    let next_buffer = &mut self.buffer[self.leftover..];
    Ok(NextBuffer {
      ptr: next_buffer.as_mut_ptr() as u32,
      size: next_buffer.len() as u32,
    })
  }
  /* [...] */
}

let componentCore = 'component.core.wasm'
let componentCoreOpt = 'component.core.opt.wasm'

// [...]

await appendFile(`${out}/component.d.ts`, ';export const memory0: WebAssembly.Memory;')

await modifyFile(`${out}/component.js`, (contents) => {
  let edited = contents.replace(
    `new URL('./${componentCore}', import.meta.url)`,
    (x) =>
      `(import.meta.env.DEV ? ${x} : ${x.replace(componentCore, componentCoreOpt)})`,
  )

  assert(contents !== edited)
  return edited + ';export {memory0};'
})

await $`../node_modules/binaryen/bin/wasm-opt \
  --strip-debug -o ${out}/${componentCoreOpt} ${out}/${componentCore}`

// original:
new URL('./component.core.wasm')
// changed:
(import.meta.env.DEV
  ? new URL('./component.core.wasm')
  : new URL('./component.core.opt.wasm'))

let buffer = rd instanceof ReadableStreamDefaultReader
  ? null
  : new ArrayBuffer(bufferSize)
let mem = new Uint8Array(impl.memory0.buffer)

let loader = new impl.loader.Loader(buffer.byteLength)
let { ptr: targetPtr, size: targetSize } = loader.processBuffer(0)

while (true) {
  if (buffer !== null && targetSize > buffer.byteLength) {
    buffer = new ArrayBuffer(targetSize)
  }

  let { value, done } = rd instanceof ReadableStreamDefaultReader
    ? await rd.read()
    : await rd.read(new Uint8Array(buffer!, 0, targetSize))
  if (value === undefined) {
    break
  }

  let offset = 0
  while (offset < value.length) {
    let remaining = value.length - offset
    let toSend = offset === 0 && remaining <= targetSize
      ? value
      : value.slice(offset, Math.min(targetSize, remaining))

    if (mem.byteLength === 0) {
      mem = new Uint8Array(impl.memory0.buffer)
    }
    mem.set(toSend, targetPtr)
    ;({ ptr: targetPtr, size: targetSize } = loader.processBuffer(toSend.byteLength))

    offset += toSend.byteLength
  }

  totalProcessed += value.byteLength

  if (done) {
    break
  }
  buffer = value.buffer
}

try {
  byobReader = stream.getReader({ mode: 'byob' })
} catch (ex) {
  if (!(ex instanceof TypeError)) {
    throw ex
  }
}

// [...]
return await loadFromReader(
  byobReader ?? stream.getReader(),
  progress,
  options,
);

while (offset < value.length) {
  let remaining = value.length - offset
  let toSend = offset === 0 && remaining <= targetSize
    ? value
    : value.slice(offset, Math.min(targetSize, remaining))
  // [...]
  offset += toSend.byteLength
}

if (mem.byteLength === 0) {
  mem = new Uint8Array(impl.memory0.buffer)
}

static already_AddRefed<nsICloneableInputStream> EnsureCloneableStream(
    nsIInputStream* aInputStream, uint64_t aLength) {
  // [...]

  // If the stream we're copying is known to be small, specify the size of the
  // pipe's segments precisely to limit wasted space. An extra byte above length
  // is included to avoid allocating an extra segment immediately before reading
  // EOF from the source stream.  Otherwise, allocate 64k buffers rather than
  // the default of 4k buffers to reduce segment counts for very large payloads.
  static constexpr uint32_t kBaseSegmentSize = 64 * 1024;
  uint32_t segmentSize = kBaseSegmentSize;
  if (aLength + 1 <= kBaseSegmentSize * 4) {
    segmentSize = aLength + 1;
  }

  // NOTE: We specify unlimited segments to eagerly build a complete copy of the
  // source stream locally without waiting for the blob to be consumed.
  nsCOMPtr<nsIAsyncInputStream> reader;
  nsCOMPtr<nsIAsyncOutputStream> writer;
  NS_NewPipe2(getter_AddRefs(reader), getter_AddRefs(writer), true, true,
              segmentSize, UINT32_MAX);

  // [...]
  cloneable = do_QueryInterface(reader);
  MOZ_ASSERT(cloneable && cloneable->GetCloneable());
  return cloneable.forget();
}

// [...]
/* static */
already_AddRefed<StreamBlobImpl> StreamBlobImpl::Create(
    already_AddRefed<nsIInputStream> aInputStream, const nsAString& aName,
    const nsAString& aContentType, int64_t aLastModifiedDate, uint64_t aLength,
    const nsAString& aBlobImplType) {
  nsCOMPtr<nsIInputStream> inputStream = std::move(aInputStream);
  nsCOMPtr<nsICloneableInputStream> cloneable =
      EnsureCloneableStream(inputStream, aLength);

  RefPtr<StreamBlobImpl> blobImplStream =
      new StreamBlobImpl(cloneable.forget(), aName, aContentType,
                         aLastModifiedDate, aLength, aBlobImplType);
  blobImplStream->MaybeRegisterMemoryReporter();
  return blobImplStream.forget();
}

BytesConsumer::Result BlobBytesConsumer::BeginRead(const char** buffer,
                                                   size_t* available) {
  if (!nested_consumer_) {
    if (!blob_data_handle_)
      return Result::kDone;

    // Create a DataPipe to transport the data from the blob.
    MojoCreateDataPipeOptions options;
    options.struct_size = sizeof(MojoCreateDataPipeOptions);
    options.flags = MOJO_CREATE_DATA_PIPE_FLAG_NONE;
    options.element_num_bytes = 1;
    options.capacity_num_bytes =
        blink::BlobUtils::GetDataPipeCapacity(blob_data_handle_->size());

    // [...]
  }
  return nested_consumer_->BeginRead(buffer, available);
}

// The 2MB limit was selected via a finch trial.
constexpr int kBlobDefaultDataPipeCapacity = 2 * 1024 * 1024;

constexpr base::FeatureParam<int> kBlobDataPipeCapacity{
    &kBlobDataPipeTuningFeature, "capacity_bytes",
    kBlobDefaultDataPipeCapacity};

// [...]

// static
uint32_t BlobUtils::GetDataPipeCapacity(uint64_t target_blob_size) {
  static_assert(kUnknownSize > kBlobDefaultDataPipeCapacity,
                "The unknown size constant must be greater than our capacity.");
  uint32_t result =
      std::min(base::saturated_cast<uint32_t>(target_blob_size),
               base::saturated_cast<uint32_t>(kBlobDataPipeCapacity.Get()));
  return std::max(result,
                  base::saturated_cast<uint32_t>(kBlobMinDataPipeCapacity));
}

static const unsigned bufferSize = 512 * 1024;

// [...]

client()->didReceiveResponseAsync(this, WTFMove(response), [this, protectedThis = Ref { *this }] {
    m_buffer.resize(bufferSize);
    readAsync();
});