Something went wrong. Try again.
PLC Bundle V1 Example Implementations
Something went wrong. Try again.
5.6 kB · 185 lines
Ruby
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186#!/usr/bin/env ruby# frozen_string_literal: true
# plcbundle.rb - Ruby implementation of plcbundle V1 specification# Creates compressed, cryptographically-chained bundles of DID PLC operations## PLC Bundle v1 Specification:# https://tangled.org/atscan.net/plcbundle/blob/main/docs/specification.md
require 'json'require 'digest'require 'net/http'require 'uri'require 'fileutils'require 'time'require 'set'require 'zstd-ruby'
# Configuration constantsBUNDLE_SIZE = 10_000INDEX_FILE = 'plc_bundles.json'PLC_URL = 'https://plc.directory'
class PlcBundle def initialize(dir) @dir = dir @pool = [] # Mempool of operations waiting to be bundled @seen = Set.new # CID deduplication set (pruned after each bundle) FileUtils.mkdir_p(@dir) @idx = load_idx puts "plcbundle v1 | Dir: #{@dir} | Last: #{@idx[:last_bundle]}\n" # Seed deduplication set with boundary CIDs from previous bundle seed_boundary if @idx[:bundles].any? end
def run cursor = @idx[:bundles].last&.dig(:end_time) loop do puts "\nFetch: #{cursor || 'start'}" ops = fetch(cursor) or (puts('Done.') and break) add_ops(ops) # Validate and add to mempool cursor = ops.last[:time] create_bundle while @pool.size >= BUNDLE_SIZE # Create bundles when ready sleep 0.2 # Rate limiting end save_idx puts "\nBundles: #{@idx[:bundles].size} | Pool: #{@pool.size} | Size: #{'%.1f' % (@idx[:total_size_bytes] / 1e6)}MB" rescue => e puts "\nError: #{e.message}" and save_idx end
private
# Fetch operations from PLC directory export endpoint def fetch(after) uri = URI("#{PLC_URL}/export?count=1000#{after ? "&after=#{after}" : ''}") res = Net::HTTP.get_response(uri) res.is_a?(Net::HTTPSuccess) or return nil # Parse each line and preserve raw JSON for reproducibility (Spec 4.2) res.body.strip.split("\n").map do |line| {**JSON.parse(line, symbolize_names: true), raw: line, time: JSON.parse(line)['createdAt']} end rescue nil end
# Process and validate operations before adding to mempool def add_ops(ops) last_t = @pool.last&.dig(:time) || @idx[:bundles].last&.dig(:end_time) || '' added = 0 ops.each do |op| next if @seen.include?(op[:cid]) # Skip duplicates (boundary + within-batch) # Spec 3: Validate chronological order raise "Order fail" if op[:time] < last_t @pool << op @seen << op[:cid] last_t = op[:time] added += 1 end puts " +#{added} ops" end
# Create a bundle file and update index def create_bundle ops = @pool.shift(BUNDLE_SIZE) parent = @idx[:bundles].last&.dig(:hash) || '' # Spec 4.2: Serialize using raw JSON strings for reproducibility jsonl = ops.map { |o| o[:raw] + "\n" }.join # Spec 6.3: Calculate hashes ch = sha(jsonl) # Content hash h = sha(parent.empty? ? "plcbundle:genesis:#{ch}" : "#{parent}:#{ch}") # Chain hash zst = Zstd.compress(jsonl) # Compress # Write bundle file num = @idx[:last_bundle] + 1 file = format('%06d.jsonl.zst', num) File.binwrite("#{@dir}/#{file}", zst)
# Create metadata entry @idx[:bundles] << { bundle_number: num, start_time: ops[0][:time], end_time: ops[-1][:time], operation_count: ops.size, did_count: ops.map { |o| o[:did] }.uniq.size, hash: h, content_hash: ch, parent: parent, compressed_hash: sha(zst), compressed_size: zst.bytesize, uncompressed_size: jsonl.bytesize, cursor: @idx[:bundles].last&.dig(:end_time) || '', created_at: Time.now.utc.iso8601 } @idx[:last_bundle] = num @idx[:total_size_bytes] += zst.bytesize # Prune seen CIDs: only keep boundary + mempool (memory efficient) @seen = boundary_cids(ops) | @pool.map { |o| o[:cid] }.to_set save_idx puts "✓ #{file} | #{h[0..12]}... | seen:#{@seen.size}" end
# Load index from disk or create new def load_idx JSON.parse(File.read("#{@dir}/#{INDEX_FILE}"), symbolize_names: true) rescue {version: '1.0', last_bundle: 0, updated_at: '', total_size_bytes: 0, bundles: []} end
# Atomically save index using temp file + rename def save_idx @idx[:updated_at] = Time.now.utc.iso8601 tmp = "#{@dir}/#{INDEX_FILE}.tmp" File.write(tmp, JSON.pretty_generate(@idx)) File.rename(tmp, "#{@dir}/#{INDEX_FILE}") end
# Seed deduplication set with CIDs from last bundle's boundary def seed_boundary last = @idx[:bundles].last file = format('%06d.jsonl.zst', last[:bundle_number]) data = Zstd.decompress(File.binread("#{@dir}/#{file}")) ops = data.strip.split("\n").map do |line| {time: JSON.parse(line)['createdAt'], cid: JSON.parse(line)['cid']} end @seen = boundary_cids(ops) puts "Seeded: #{@seen.size} CIDs from bundle #{last[:bundle_number]}" rescue puts "Warning: couldn't seed boundary" end
# Get CIDs from operations at the same timestamp as the last op (boundary) def boundary_cids(ops) return Set.new if ops.empty? t = ops[-1][:time] ops.reverse.take_while { |o| o[:time] == t }.map { |o| o[:cid] }.to_set end
# SHA-256 hash helper def sha(data) Digest::SHA256.hexdigest(data) endend
# Entry pointPlcBundle.new(ARGV[0] || './plc_bundles_rb').run if __FILE__ == $PROGRAM_NAME