v2.2.0

feat(storage): add Bitcask storage migration, binary WAL, and data compaction support
v2.1.1
2026-04-04 19:49:47 +00:00 · 2026-04-04 19:49:47 +00:00 · 2026-04-02 17:05:07 +00:00 · 2026-04-02 17:05:07 +00:00 · 2026-04-02 17:02:03 +00:00 · 2026-04-02 17:02:03 +00:00
41 changed files with 6347 additions and 475 deletions
@@ -32,6 +32,17 @@
  "@git.zone/tsdoc": {
    "legal": "\n## License and Legal Information\n\nThis repository contains open-source code that is licensed under the MIT License. A copy of the MIT License can be found in the [license](license) file within this repository. \n\n**Please note:** The MIT License does not grant permission to use the trade names, trademarks, service marks, or product names of the project, except as required for reasonable and customary use in describing the origin of the work and reproducing the content of the NOTICE file.\n\n### Trademarks\n\nThis project is owned and maintained by Task Venture Capital GmbH. The names and logos associated with Task Venture Capital GmbH and any related products or services are trademarks of Task Venture Capital GmbH and are not included within the scope of the MIT license granted herein. Use of these trademarks must comply with Task Venture Capital GmbH's Trademark Guidelines, and any usage must be approved in writing by Task Venture Capital GmbH.\n\n### Company Information\n\nTask Venture Capital GmbH  \nRegistered at District court Bremen HRB 35230 HB, Germany\n\nFor any legal inquiries or if you require further information, please contact us via email at hello@task.vc.\n\nBy using this repository, you acknowledge that you have read this section, agree to comply with its terms, and understand that the licensing of the code does not imply endorsement by Task Venture Capital GmbH of any derivative works.\n"
  },
+  "@git.zone/tsbundle": {
+    "bundles": [
+      {
+        "from": "./ts_debugui/index.ts",
+        "to": "./ts_debugserver/bundled.ts",
+        "outputMode": "base64ts",
+        "bundler": "esbuild",
+        "includeFiles": ["./html/index.html"]
+      }
+    ]
+  },
  "@git.zone/tsrust": {
    "targets": ["linux_amd64", "linux_arm64"]
  },
@@ -1,5 +1,24 @@
 # Changelog

+## 2026-04-04 - 2.2.0 - feat(storage)
+add Bitcask storage migration, binary WAL, and data compaction support
+
+- add TypeScript storage migration from legacy JSON collections to the v1 Bitcask binary format before starting the Rust engine
+- replace the legacy JSON WAL with a binary write-ahead log plus shared binary record and KeyDir infrastructure in rustdb-storage
+- introduce data file compaction with dead-record reclamation and tests, and add the bson dependency for BSON serialization during migration
+
+## 2026-04-02 - 2.1.1 - fix(package)
+update package metadata
+
+- Adjusts a single package metadata entry in package.json.
+
+## 2026-04-02 - 2.1.0 - feat(smartdb)
+add operation log APIs, point-in-time revert support, and a web-based debug dashboard
+
+- records insert, update, and delete operations with before/after document snapshots in the Rust oplog
+- adds management and TypeScript APIs for metrics, oplog queries, collection browsing, document browsing, and revert-to-sequence operations
+- introduces new debugserver and debugui package exports with bundled browser assets served through typedserver
+
 ## 2026-03-26 - 2.0.0 - BREAKING CHANGE(core)
 replace the TypeScript database engine with a Rust-backed embedded server and bridge

@@ -0,0 +1,13 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+  <meta charset="UTF-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+  <title>SmartDB Debug</title>
+  <style>body { margin: 0; background: #09090b; }</style>
+  <script type="module" src="/bundle.js"></script>
+</head>
+<body>
+  <smartdb-debugui></smartdb-debugui>
+</body>
+</html>
@@ -1,10 +1,12 @@
 {
  "name": "@push.rocks/smartdb",
-  "version": "2.0.0",
+  "version": "2.2.0",
  "private": false,
  "description": "A MongoDB-compatible embedded database server with wire protocol support, backed by a high-performance Rust engine.",
  "exports": {
-    ".": "./dist_ts/index.js"
+    ".": "./dist_ts/index.js",
+    "./debugui": "./dist_ts_debugui/index.js",
+    "./debugserver": "./dist_ts_debugserver/index.js"
  },
  "type": "module",
  "author": "Task Venture Capital GmbH",
@@ -12,7 +14,7 @@
  "scripts": {
    "test:before": "(tsrust)",
    "test": "(tstest test/. --verbose --logfile --timeout 60)",
-    "build": "(tsbuild tsfolders) && (tsrust)",
+    "build": "(tsbundle) && (tsbuild tsfolders) && (tsrust)",
    "buildDocs": "tsdoc"
  },
  "devDependencies": {
@@ -25,7 +27,10 @@
    "mongodb": "^7.1.1"
  },
  "dependencies": {
-    "@push.rocks/smartrust": "^1.3.2"
+    "@api.global/typedserver": "^8.0.0",
+    "@design.estate/dees-element": "^2.0.0",
+    "@push.rocks/smartrust": "^1.3.2",
+    "bson": "^7.2.0"
  },
  "browserslist": [
    "last 1 chrome versions"
@@ -1,6 +1,6 @@
 # @push.rocks/smartdb

-A MongoDB-compatible embedded database server powered by Rust 🦀⚡ — use the official `mongodb` driver and it just works. No binary downloads, instant startup, zero config.
+A MongoDB-compatible embedded database server powered by Rust 🦀⚡ — use the official `mongodb` driver and it just works. No binary downloads, instant startup, zero config. Features a built-in **operation log** with **point-in-time revert** and a web-based **debug dashboard**.

 ## Install

@@ -14,6 +14,8 @@ npm install @push.rocks/smartdb

 For reporting bugs, issues, or security vulnerabilities, please visit [community.foss.global/](https://community.foss.global/). This is the central community hub for all issue reporting. Developers who sign and comply with our contribution agreement and go through identification can also get a [code.foss.global/](https://code.foss.global/) account to submit Pull Requests directly.

+---
+
 ## What It Does

 `@push.rocks/smartdb` is a **real database server** that speaks the wire protocol used by MongoDB drivers. The core engine is written in Rust for high performance, with a thin TypeScript orchestration layer. Connect with the standard `mongodb` Node.js driver — no mocks, no stubs, no external binaries required.
@@ -26,52 +28,55 @@ For reporting bugs, issues, or security vulnerabilities, please visit [community
 | **Binary download** | Bundled (~7MB) | ~200MB+ |
 | **Install** | `pnpm add` | System package / Docker |
 | **Persistence** | Memory or file-based | Full disk engine |
+| **Debug UI** | Built-in 🖥️ | External tooling |
+| **Point-in-time revert** | Built-in ⏪ | Requires oplog tailing |
 | **Perfect for** | Unit tests, CI/CD, prototyping, local dev, embedded | Production at scale |

-### Two Ways to Use It
+### Three Ways to Use It

- 🏗️ **`SmartdbServer`** — Full control. Configure port, host, storage backend, Unix sockets. Great for test fixtures or custom setups.
 - 🎯 **`LocalSmartDb`** — Zero-config convenience. Give it a folder path, get a persistent database over a Unix socket. Done.
+- 🏗️ **`SmartdbServer`** — Full control. Configure port, host, storage backend, Unix sockets. Great for test fixtures or custom setups.
+- 🖥️ **`SmartdbDebugServer`** — Launch a web dashboard to visually browse collections, inspect the operation log, and revert to any point in time.

 ### Architecture: TypeScript + Rust 🦀

-SmartDB uses the same **sidecar binary** pattern as [@push.rocks/smartproxy](https://code.foss.global/push.rocks/smartproxy):
+SmartDB uses a **sidecar binary** pattern — TypeScript handles lifecycle, Rust handles all database operations:

 ```
-┌──────────────────────────────────────────────────────────┐
-│                   Your Application                        │
-│                  (TypeScript / Node.js)                    │
-│  ┌─────────────────┐     ┌───────────────────────────┐   │
-│  │  SmartdbServer   │────▶│  RustDbBridge (IPC)       │   │
-│  │  or LocalSmartDb │     │  @push.rocks/smartrust    │   │
-│  └─────────────────┘     └───────────┬───────────────┘   │
-└──────────────────────────────────────┼───────────────────┘
+┌──────────────────────────────────────────────────────────────┐
+│                   Your Application                            │
+│                  (TypeScript / Node.js)                        │
+│  ┌─────────────────┐     ┌───────────────────────────┐       │
+│  │  SmartdbServer   │────▶│  RustDbBridge (IPC)       │       │
+│  │  or LocalSmartDb │     │  @push.rocks/smartrust    │       │
+│  └─────────────────┘     └───────────┬───────────────┘       │
+└──────────────────────────────────────┼───────────────────────┘
                                       │ spawn + JSON IPC
                                       ▼
-┌──────────────────────────────────────────────────────────┐
-│                    rustdb binary 🦀                       │
-│                                                          │
-│  ┌──────────────┐  ┌──────────────┐  ┌───────────────┐  │
-│  │ Wire Protocol│→ │Command Router│→ │   Handlers    │  │
-│  │  (OP_MSG)    │  │  (40 cmds)   │  │ Find,Insert.. │  │
-│  └──────────────┘  └──────────────┘  └───────┬───────┘  │
-│                                               │          │
-│  ┌─────────┐ ┌────────┐ ┌───────────┐ ┌──────┴──────┐  │
-│  │  Query  │ │ Update │ │Aggregation│ │   Index     │  │
-│  │ Matcher │ │ Engine │ │  Engine   │ │   Engine    │  │
-│  └─────────┘ └────────┘ └───────────┘ └─────────────┘  │
-│                                                          │
-│  ┌──────────────────┐  ┌──────────────────┐             │
-│  │  MemoryStorage   │  │   FileStorage    │             │
-│  └──────────────────┘  └──────────────────┘             │
-└──────────────────────────────────────────────────────────┘
+┌──────────────────────────────────────────────────────────────┐
+│                    rustdb binary 🦀                           │
+│                                                              │
+│  ┌──────────────┐  ┌──────────────┐  ┌───────────────┐      │
+│  │ Wire Protocol│→ │Command Router│→ │   Handlers    │      │
+│  │  (OP_MSG)    │  │  (40+ cmds)  │  │ Find,Insert.. │      │
+│  └──────────────┘  └──────────────┘  └───────┬───────┘      │
+│                                               │              │
+│  ┌─────────┐ ┌────────┐ ┌───────────┐ ┌──────┴──────┐      │
+│  │  Query  │ │ Update │ │Aggregation│ │   Index     │      │
+│  │ Matcher │ │ Engine │ │  Engine   │ │   Engine    │      │
+│  └─────────┘ └────────┘ └───────────┘ └─────────────┘      │
+│                                                              │
+│  ┌──────────────────┐  ┌──────────────────┐  ┌──────────┐   │
+│  │  MemoryStorage   │  │   FileStorage    │  │  OpLog   │   │
+│  └──────────────────┘  └──────────────────┘  └──────────┘   │
+└──────────────────────────────────────────────────────────────┘
              ▲
              │ TCP / Unix Socket (wire protocol)
              │
-┌─────────────┴────────────────────────────────────────────┐
-│              MongoClient (mongodb npm driver)             │
-│              Connects directly to Rust binary             │
-└──────────────────────────────────────────────────────────┘
+┌─────────────┴────────────────────────────────────────────────┐
+│              MongoClient (mongodb npm driver)                  │
+│              Connects directly to Rust binary                 │
+└──────────────────────────────────────────────────────────────┘
 ```

 The TypeScript layer handles **lifecycle only** (start/stop/configure via IPC). All database operations flow directly from the `MongoClient` to the Rust binary over TCP or Unix sockets — **zero per-query IPC overhead**.
@@ -128,6 +133,87 @@ await client.close();
 await server.stop();
 ```

+### Option 3: Debug Server (Visual Dashboard) 🖥️
+
+Launch a web-based dashboard to inspect your database in real time:
+
+```typescript
+import { SmartdbServer } from '@push.rocks/smartdb';
+import { SmartdbDebugServer } from '@push.rocks/smartdb/debugserver';
+
+const server = new SmartdbServer({ storage: 'memory' });
+await server.start();
+
+const debugServer = new SmartdbDebugServer(server, { port: 4000 });
+await debugServer.start();
+// Open http://localhost:4000 in your browser 🚀
+```
+
+The debug dashboard gives you:
+- 📊 **Dashboard** — server status, uptime, database/collection counts, operation breakdown
+- 📁 **Collection Browser** — browse databases, collections, and documents interactively
+- 📝 **OpLog Timeline** — every insert, update, and delete with expandable field-level diffs
+- ⏪ **Point-in-Time Revert** — select any oplog sequence, preview what will be undone, and execute
+
+---
+
+## 📝 Operation Log & Point-in-Time Revert
+
+Every write operation (insert, update, delete) is automatically recorded in an in-memory **operation log (OpLog)** with full before/after document snapshots. This enables:
+
+- **Change tracking** — see exactly what changed, when, and in which collection
+- **Field-level diffs** — compare previous and new document states
+- **Point-in-time revert** — undo operations back to any sequence number
+- **Dry-run preview** — see what would be reverted before executing
+
+### Programmatic OpLog API
+
+```typescript
+import { SmartdbServer } from '@push.rocks/smartdb';
+
+const server = new SmartdbServer({ port: 27017 });
+await server.start();
+
+// ... perform some CRUD operations via MongoClient ...
+
+// Get oplog entries
+const oplog = await server.getOpLog({ limit: 50 });
+console.log(oplog.entries);
+// [{ seq: 1, op: 'insert', db: 'myapp', collection: 'users', document: {...}, previousDocument: null }, ...]
+
+// Get aggregate stats
+const stats = await server.getOpLogStats();
+console.log(stats);
+// { currentSeq: 42, totalEntries: 42, entriesByOp: { insert: 20, update: 15, delete: 7 } }
+
+// Preview a revert (dry run)
+const preview = await server.revertToSeq(30, true);
+console.log(`Would undo ${preview.reverted} operations`);
+
+// Execute the revert — undoes all operations after seq 30
+const result = await server.revertToSeq(30, false);
+console.log(`Reverted ${result.reverted} operations`);
+
+// Browse collections programmatically
+const collections = await server.getCollections();
+const docs = await server.getDocuments('myapp', 'users', 50, 0);
+```
+
+### OpLog Entry Structure
+
+Each entry contains:
+
+| Field | Type | Description |
+|---|---|---|
+| `seq` | `number` | Monotonically increasing sequence number |
+| `timestampMs` | `number` | Unix timestamp in milliseconds |
+| `op` | `'insert' \| 'update' \| 'delete'` | Operation type |
+| `db` | `string` | Database name |
+| `collection` | `string` | Collection name |
+| `documentId` | `string` | Document `_id` as hex string |
+| `document` | `object \| null` | New document state (null for deletes) |
+| `previousDocument` | `object \| null` | Previous document state (null for inserts) |
+
 ---

 ## API Reference
@@ -175,6 +261,12 @@ const server = new SmartdbServer({
 | `port` | `number` | Configured port (TCP mode) |
 | `host` | `string` | Configured host (TCP mode) |
 | `socketPath` | `string \| undefined` | Socket path (socket mode) |
+| `getMetrics()` | `Promise<ISmartDbMetrics>` | Server metrics (db/collection counts, uptime) |
+| `getOpLog(params?)` | `Promise<IOpLogResult>` | Query oplog entries with optional filters |
+| `getOpLogStats()` | `Promise<IOpLogStats>` | Aggregate oplog statistics |
+| `revertToSeq(seq, dryRun?)` | `Promise<IRevertResult>` | Revert to a specific oplog sequence |
+| `getCollections(db?)` | `Promise<ICollectionInfo[]>` | List all collections with counts |
+| `getDocuments(db, coll, limit?, skip?)` | `Promise<IDocumentsResult>` | Browse documents with pagination |

 ### LocalSmartDb

@@ -202,13 +294,34 @@ const db = new LocalSmartDb({
 | `getServer()` | `SmartdbServer` | Access the underlying server |
 | `running` | `boolean` | Whether the server is running |

-#### Connection Info (`ILocalSmartDbConnectionInfo`)
+### SmartdbDebugServer
+
+Web-based debug dashboard served via `@api.global/typedserver`. Import from the `debugserver` subpath:

 ```typescript
-interface ILocalSmartDbConnectionInfo {
-  socketPath: string;      // e.g., /tmp/smartdb-abc123.sock
-  connectionUri: string;   // e.g., mongodb://%2Ftmp%2Fsmartdb-abc123.sock
-}
+import { SmartdbDebugServer } from '@push.rocks/smartdb/debugserver';
+
+const debugServer = new SmartdbDebugServer(server, { port: 4000 });
+await debugServer.start();
+// Dashboard at http://localhost:4000
+
+await debugServer.stop();
+```
+
+The UI is bundled as base64-encoded content (via `@git.zone/tsbundle`) and served from memory — no static file directory needed.
+
+### SmartdbDebugUi (Web Component)
+
+For embedding the debug UI directly into your own web application, import the `<smartdb-debugui>` web component:
+
+```typescript
+import { SmartdbDebugUi } from '@push.rocks/smartdb/debugui';
+
+// In your HTML/lit template:
+// <smartdb-debugui .server=${mySmartdbServer}></smartdb-debugui>
+//
+// Or in HTTP mode (when served by SmartdbDebugServer):
+// <smartdb-debugui apiBaseUrl=""></smartdb-debugui>
 ```

 ---
@@ -377,10 +490,10 @@ The Rust engine is organized as a Cargo workspace with 8 focused crates:
 | `rustdb-config` | Server configuration types (serde, camelCase JSON) |
 | `rustdb-wire` | Wire protocol parser/encoder (OP_MSG, OP_QUERY, OP_REPLY) |
 | `rustdb-query` | Query matcher, update engine, aggregation, sort, projection |
-| `rustdb-storage` | Storage backends (memory, file) + WAL + OpLog |
+| `rustdb-storage` | Storage backends (memory, file), OpLog with point-in-time replay |
 | `rustdb-index` | B-tree/hash indexes, query planner (IXSCAN/COLLSCAN) |
 | `rustdb-txn` | Transaction + session management with snapshot isolation |
-| `rustdb-commands` | 40 command handlers wiring everything together |
+| `rustdb-commands` | 40+ command handlers wiring everything together |

 Cross-compiled for `linux_amd64` and `linux_arm64` via [@git.zone/tsrust](https://www.npmjs.com/package/@git.zone/tsrust).

@@ -410,6 +523,12 @@ tap.test('should insert and find', async () => {
  expect(item?.price).toEqual(9.99);
 });

+tap.test('should track changes in oplog', async () => {
+  const oplog = await server.getOpLog();
+  expect(oplog.entries.length).toBeGreaterThan(0);
+  expect(oplog.entries[0].op).toEqual('insert');
+});
+
 tap.test('teardown', async () => {
  await client.close();
  await server.stop();
@@ -422,7 +541,7 @@ export default tap.start();

 ## License and Legal Information

-This repository contains open-source code licensed under the MIT License. A copy of the license can be found in the [license](./license) file.
+This repository contains open-source code licensed under the MIT License. A copy of the license can be found in the [LICENSE](./LICENSE) file.

 **Please note:** The MIT License does not grant permission to use the trade names, trademarks, service marks, or product names of the project, except as required for reasonable and customary use in describing the origin of the work and reproducing the content of the NOTICE file.

@@ -434,7 +553,7 @@ Use of these trademarks must comply with Task Venture Capital GmbH's Trademark G

 ### Company Information

-Task Venture Capital GmbH
+Task Venture Capital GmbH  
 Registered at District Court Bremen HRB 35230 HB, Germany

 For any legal inquiries or further information, please contact us via email at hello@task.vc.
@@ -275,6 +275,12 @@ dependencies = [
 "windows-sys",
 ]

+[[package]]
+name = "fastrand"
+version = "2.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a043dc74da1e37d6afe657061213aa6f425f855399a11d3463c6ecccc4dfda1f"
+
 [[package]]
 name = "find-msvc-tools"
 version = "0.1.9"
@@ -477,6 +483,12 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "linux-raw-sys"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"
+
 [[package]]
 name = "lock_api"
 version = "0.4.14"
@@ -802,6 +814,7 @@ dependencies = [
 "dashmap",
 "serde",
 "serde_json",
+ "tempfile",
 "thiserror",
 "tokio",
 "tracing",
@@ -835,6 +848,19 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "rustix"
+version = "1.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
+dependencies = [
+ "bitflags",
+ "errno",
+ "libc",
+ "linux-raw-sys",
+ "windows-sys",
+]
+
 [[package]]
 name = "rustversion"
 version = "1.0.22"
@@ -977,6 +1003,19 @@ version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369"

+[[package]]
+name = "tempfile"
+version = "3.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
+dependencies = [
+ "fastrand",
+ "getrandom 0.4.2",
+ "once_cell",
+ "rustix",
+ "windows-sys",
+]
+
 [[package]]
 name = "thiserror"
 version = "2.0.18"
@@ -66,6 +66,9 @@ uuid = { version = "1", features = ["v4", "serde"] }
 # Async traits
 async-trait = "0.1"

+# Test utilities
+tempfile = "3"
+
 # Internal crates
 rustdb-config = { path = "crates/rustdb-config" }
 rustdb-wire = { path = "crates/rustdb-wire" }
@@ -3,7 +3,7 @@ use std::sync::Arc;
 use bson::Document;
 use dashmap::DashMap;
 use rustdb_index::IndexEngine;
-use rustdb_storage::StorageAdapter;
+use rustdb_storage::{OpLog, StorageAdapter};
 use rustdb_txn::{SessionEngine, TransactionEngine};

 /// Shared command execution context, passed to all handlers.
@@ -20,6 +20,8 @@ pub struct CommandContext {
    pub cursors: Arc<DashMap<i64, CursorState>>,
    /// Server start time (for uptime reporting).
    pub start_time: std::time::Instant,
+    /// Operation log for point-in-time replay.
+    pub oplog: Arc<OpLog>,
 }

 /// State of an open cursor from a find or aggregate command.
@@ -2,6 +2,7 @@ use std::collections::HashSet;

 use bson::{doc, Bson, Document};
 use rustdb_query::QueryMatcher;
+use rustdb_storage::OpType;
 use tracing::debug;

 use crate::context::CommandContext;
@@ -171,6 +172,16 @@ async fn delete_matching(
            .await
            .map_err(|e| CommandError::StorageError(e.to_string()))?;

+        // Record in oplog.
+        ctx.oplog.append(
+            OpType::Delete,
+            db,
+            coll,
+            &id_str,
+            None,
+            Some(doc.clone()),
+        );
+
        // Update index engine.
        if let Some(mut engine) = ctx.indexes.get_mut(ns_key) {
            engine.on_delete(doc);
@@ -2,6 +2,7 @@ use std::collections::HashMap;

 use bson::{doc, oid::ObjectId, Bson, Document};
 use rustdb_index::IndexEngine;
+use rustdb_storage::OpType;
 use tracing::{debug, warn};

 use crate::context::CommandContext;
@@ -63,7 +64,17 @@ pub async fn handle(

        // Attempt storage insert.
        match ctx.storage.insert_one(db, coll, doc.clone()).await {
-            Ok(_id_str) => {
+            Ok(id_str) => {
+                // Record in oplog.
+                ctx.oplog.append(
+                    OpType::Insert,
+                    db,
+                    coll,
+                    &id_str,
+                    Some(doc.clone()),
+                    None,
+                );
+
                // Update index engine.
                let mut engine = ctx
                    .indexes
@@ -3,6 +3,7 @@ use std::collections::HashSet;
 use bson::{doc, oid::ObjectId, Bson, Document};
 use rustdb_index::IndexEngine;
 use rustdb_query::{QueryMatcher, UpdateEngine, sort_documents, apply_projection};
+use rustdb_storage::OpType;
 use tracing::debug;

 use crate::context::CommandContext;
@@ -151,7 +152,17 @@ async fn handle_update(

                    // Insert the new document.
                    match ctx.storage.insert_one(db, coll, updated.clone()).await {
-                        Ok(_) => {
+                        Ok(id_str) => {
+                            // Record upsert in oplog as an insert.
+                            ctx.oplog.append(
+                                OpType::Insert,
+                                db,
+                                coll,
+                                &id_str,
+                                Some(updated.clone()),
+                                None,
+                            );
+
                            // Update index.
                            let mut engine = ctx
                                .indexes
@@ -212,6 +223,16 @@ async fn handle_update(
                            .await
                        {
                            Ok(()) => {
+                                // Record in oplog.
+                                ctx.oplog.append(
+                                    OpType::Update,
+                                    db,
+                                    coll,
+                                    &id_str,
+                                    Some(updated_doc.clone()),
+                                    Some(matched_doc.clone()),
+                                );
+
                                // Update index.
                                if let Some(mut engine) = ctx.indexes.get_mut(&ns_key) {
                                    let _ = engine.on_update(matched_doc, &updated_doc);
@@ -362,6 +383,16 @@ async fn handle_find_and_modify(
            let id_str = extract_id_string(doc);
            ctx.storage.delete_by_id(db, coll, &id_str).await?;

+            // Record in oplog.
+            ctx.oplog.append(
+                OpType::Delete,
+                db,
+                coll,
+                &id_str,
+                None,
+                Some(doc.clone()),
+            );
+
            // Update index.
            if let Some(mut engine) = ctx.indexes.get_mut(&ns_key) {
                engine.on_delete(doc);
@@ -418,6 +449,16 @@ async fn handle_find_and_modify(
            .update_by_id(db, coll, &id_str, updated_doc.clone())
            .await?;

+        // Record in oplog.
+        ctx.oplog.append(
+            OpType::Update,
+            db,
+            coll,
+            &id_str,
+            Some(updated_doc.clone()),
+            Some(original_doc.clone()),
+        );
+
        // Update index.
        if let Some(mut engine) = ctx.indexes.get_mut(&ns_key) {
            let _ = engine.on_update(&original_doc, &updated_doc);
@@ -464,10 +505,20 @@ async fn handle_find_and_modify(
            updated_doc.get("_id").unwrap().clone()
        };

-        ctx.storage
+        let inserted_id_str = ctx.storage
            .insert_one(db, coll, updated_doc.clone())
            .await?;

+        // Record upsert in oplog as an insert.
+        ctx.oplog.append(
+            OpType::Insert,
+            db,
+            coll,
+            &inserted_id_str,
+            Some(updated_doc.clone()),
+            None,
+        );
+
        // Update index.
        {
            let mut engine = ctx
@@ -17,3 +17,6 @@ tracing = { workspace = true }
 crc32fast = { workspace = true }
 uuid = { workspace = true }
 async-trait = { workspace = true }
+
+[dev-dependencies]
+tempfile = { workspace = true }
@@ -0,0 +1,499 @@
+//! Binary Write-Ahead Log for crash recovery.
+//!
+//! # Protocol
+//!
+//! Every mutation follows this sequence:
+//! 1. Append WAL record → fsync
+//! 2. Perform the actual data write
+//! 3. Append WAL commit marker → fsync
+//!
+//! On recovery, uncommitted entries (those without a matching commit marker)
+//! are replayed or verified.
+//!
+//! # Record format
+//!
+//! ```text
+//! ┌──────────┬──────────┬──────────┬──────────┬──────────┬──────────┬────────────┐
+//! │ magic    │ seq      │ op       │ key_len  │ val_len  │ crc32    │ payload    │
+//! │ u16 LE   │ u64 LE   │ u8       │ u32 LE   │ u32 LE   │ u32 LE   │ [key][val] │
+//! │ 0xWA01   │          │          │          │          │          │            │
+//! └──────────┴──────────┴──────────┴──────────┴──────────┴──────────┴────────────┘
+//! ```
+//!
+//! # Commit marker
+//!
+//! ```text
+//! ┌──────────┬──────────┬──────────┐
+//! │ magic    │ seq      │ crc32    │
+//! │ u16 LE   │ u64 LE   │ u32 LE   │
+//! │ 0xCA01   │          │          │
+//! └──────────┴──────────┴──────────┘
+//! ```
+
+use std::io::{self, BufReader, Read, Write};
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use crate::error::{StorageError, StorageResult};
+use crate::record::{FileHeader, FileType, FILE_HEADER_SIZE};
+
+// ---------------------------------------------------------------------------
+// Constants
+// ---------------------------------------------------------------------------
+
+const WAL_RECORD_MAGIC: u16 = 0xAA01;
+const WAL_COMMIT_MAGIC: u16 = 0xCC01;
+
+/// WAL record header: magic(2) + seq(8) + op(1) + key_len(4) + val_len(4) + crc(4) = 23
+const WAL_RECORD_HEADER: usize = 23;
+
+/// Commit marker size: magic(2) + seq(8) + crc(4) = 14
+const WAL_COMMIT_SIZE: usize = 14;
+
+// ---------------------------------------------------------------------------
+// WAL operation type
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum WalOpType {
+    Insert = 1,
+    Update = 2,
+    Delete = 3,
+}
+
+impl WalOpType {
+    fn from_u8(v: u8) -> StorageResult<Self> {
+        match v {
+            1 => Ok(WalOpType::Insert),
+            2 => Ok(WalOpType::Update),
+            3 => Ok(WalOpType::Delete),
+            _ => Err(StorageError::WalError(format!("unknown WAL op: {v}"))),
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// WAL entry (parsed from file)
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone)]
+pub struct WalEntry {
+    pub seq: u64,
+    pub op: WalOpType,
+    pub key: Vec<u8>,
+    pub value: Vec<u8>,
+}
+
+// ---------------------------------------------------------------------------
+// Internal: what we read from the WAL file
+// ---------------------------------------------------------------------------
+
+#[derive(Debug)]
+enum WalItem {
+    Record(WalEntry),
+    Commit(u64), // seq that was committed
+}
+
+// ---------------------------------------------------------------------------
+// BinaryWal
+// ---------------------------------------------------------------------------
+
+/// Binary write-ahead log backed by a single file.
+pub struct BinaryWal {
+    path: PathBuf,
+    next_seq: AtomicU64,
+}
+
+impl BinaryWal {
+    /// Create a new WAL. Does not touch the filesystem until `initialize()`.
+    pub fn new(path: PathBuf) -> Self {
+        Self {
+            path,
+            next_seq: AtomicU64::new(1),
+        }
+    }
+
+    /// Initialize: create parent dirs, recover sequence counter from existing file.
+    pub fn initialize(&self) -> StorageResult<()> {
+        if let Some(parent) = self.path.parent() {
+            std::fs::create_dir_all(parent)?;
+        }
+
+        if self.path.exists() {
+            // Scan to find highest seq
+            let items = self.read_all_items()?;
+            let max_seq = items
+                .iter()
+                .map(|item| match item {
+                    WalItem::Record(e) => e.seq,
+                    WalItem::Commit(s) => *s,
+                })
+                .max()
+                .unwrap_or(0);
+            self.next_seq.store(max_seq + 1, Ordering::SeqCst);
+        } else {
+            // Create the file with a header
+            let mut f = std::fs::File::create(&self.path)?;
+            let hdr = FileHeader::new(FileType::Wal);
+            f.write_all(&hdr.encode())?;
+            f.flush()?;
+            f.sync_all()?;
+        }
+
+        Ok(())
+    }
+
+    /// Append a WAL record. Returns the sequence number. Fsyncs.
+    pub fn append(
+        &self,
+        op: WalOpType,
+        key: &[u8],
+        value: &[u8],
+    ) -> StorageResult<u64> {
+        let seq = self.next_seq.fetch_add(1, Ordering::SeqCst);
+        let key_len = key.len() as u32;
+        let val_len = value.len() as u32;
+
+        // Build header bytes (without CRC)
+        let mut hdr = Vec::with_capacity(WAL_RECORD_HEADER);
+        hdr.extend_from_slice(&WAL_RECORD_MAGIC.to_le_bytes());
+        hdr.extend_from_slice(&seq.to_le_bytes());
+        hdr.push(op as u8);
+        hdr.extend_from_slice(&key_len.to_le_bytes());
+        hdr.extend_from_slice(&val_len.to_le_bytes());
+        // CRC placeholder
+        hdr.extend_from_slice(&0u32.to_le_bytes());
+
+        // Compute CRC over header (without crc field) + payload
+        let mut hasher = crc32fast::Hasher::new();
+        hasher.update(&hdr[0..19]); // magic + seq + op + key_len + val_len
+        hasher.update(key);
+        hasher.update(value);
+        let crc = hasher.finalize();
+        hdr[19..23].copy_from_slice(&crc.to_le_bytes());
+
+        // Append to file
+        let mut f = std::fs::OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(&self.path)?;
+        f.write_all(&hdr)?;
+        f.write_all(key)?;
+        f.write_all(value)?;
+        f.sync_all()?;
+
+        Ok(seq)
+    }
+
+    /// Append a commit marker for the given sequence. Fsyncs.
+    pub fn append_commit(&self, seq: u64) -> StorageResult<()> {
+        let mut buf = Vec::with_capacity(WAL_COMMIT_SIZE);
+        buf.extend_from_slice(&WAL_COMMIT_MAGIC.to_le_bytes());
+        buf.extend_from_slice(&seq.to_le_bytes());
+
+        // CRC over magic + seq
+        let mut hasher = crc32fast::Hasher::new();
+        hasher.update(&buf[0..10]);
+        let crc = hasher.finalize();
+        buf.extend_from_slice(&crc.to_le_bytes());
+
+        let mut f = std::fs::OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(&self.path)?;
+        f.write_all(&buf)?;
+        f.sync_all()?;
+
+        Ok(())
+    }
+
+    /// Recover: return all WAL entries that were NOT committed.
+    pub fn recover(&self) -> StorageResult<Vec<WalEntry>> {
+        let items = self.read_all_items()?;
+
+        // Collect committed seq numbers
+        let committed: std::collections::HashSet<u64> = items
+            .iter()
+            .filter_map(|item| {
+                if let WalItem::Commit(s) = item {
+                    Some(*s)
+                } else {
+                    None
+                }
+            })
+            .collect();
+
+        // Return records without a commit marker
+        let uncommitted: Vec<WalEntry> = items
+            .into_iter()
+            .filter_map(|item| {
+                if let WalItem::Record(entry) = item {
+                    if !committed.contains(&entry.seq) {
+                        return Some(entry);
+                    }
+                }
+                None
+            })
+            .collect();
+
+        Ok(uncommitted)
+    }
+
+    /// Truncate the WAL: rewrite with just the file header (clears all entries).
+    pub fn truncate(&self) -> StorageResult<()> {
+        let mut f = std::fs::File::create(&self.path)?;
+        let hdr = FileHeader::new(FileType::Wal);
+        f.write_all(&hdr.encode())?;
+        f.flush()?;
+        f.sync_all()?;
+        // Don't reset next_seq — it should keep incrementing
+        Ok(())
+    }
+
+    /// Path to the WAL file.
+    pub fn path(&self) -> &Path {
+        &self.path
+    }
+
+    // -----------------------------------------------------------------------
+    // Internal: read all items from the WAL file
+    // -----------------------------------------------------------------------
+
+    fn read_all_items(&self) -> StorageResult<Vec<WalItem>> {
+        if !self.path.exists() {
+            return Ok(vec![]);
+        }
+
+        let file = std::fs::File::open(&self.path)?;
+        let mut reader = BufReader::new(file);
+
+        // Skip file header (if present)
+        let file_len = std::fs::metadata(&self.path)?.len();
+        if file_len >= FILE_HEADER_SIZE as u64 {
+            let mut hdr_buf = [0u8; FILE_HEADER_SIZE];
+            reader.read_exact(&mut hdr_buf)?;
+            // Validate but don't fail hard — allow reading even slightly off headers
+            let _ = FileHeader::decode(&hdr_buf);
+        }
+
+        let mut items = Vec::new();
+
+        loop {
+            // Peek at the magic to determine if this is a record or commit marker
+            let mut magic_buf = [0u8; 2];
+            match reader.read_exact(&mut magic_buf) {
+                Ok(()) => {}
+                Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => break,
+                Err(e) => return Err(e.into()),
+            }
+            let magic = u16::from_le_bytes(magic_buf);
+
+            match magic {
+                WAL_RECORD_MAGIC => {
+                    // Read rest of header: seq(8) + op(1) + key_len(4) + val_len(4) + crc(4) = 21
+                    let mut rest = [0u8; 21];
+                    match reader.read_exact(&mut rest) {
+                        Ok(()) => {}
+                        Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => break,
+                        Err(e) => return Err(e.into()),
+                    }
+
+                    let seq = u64::from_le_bytes(rest[0..8].try_into().unwrap());
+                    let op = WalOpType::from_u8(rest[8])?;
+                    let key_len = u32::from_le_bytes(rest[9..13].try_into().unwrap()) as usize;
+                    let val_len = u32::from_le_bytes(rest[13..17].try_into().unwrap()) as usize;
+                    let stored_crc = u32::from_le_bytes(rest[17..21].try_into().unwrap());
+
+                    let mut payload = vec![0u8; key_len + val_len];
+                    match reader.read_exact(&mut payload) {
+                        Ok(()) => {}
+                        Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => break,
+                        Err(e) => return Err(e.into()),
+                    }
+
+                    // Verify CRC
+                    let mut hasher = crc32fast::Hasher::new();
+                    hasher.update(&magic_buf);
+                    hasher.update(&rest[0..17]); // seq + op + key_len + val_len
+                    hasher.update(&payload);
+                    let computed = hasher.finalize();
+
+                    if computed != stored_crc {
+                        // Corrupt WAL entry — skip it (best-effort recovery)
+                        tracing::warn!(
+                            seq,
+                            "skipping corrupt WAL record: CRC mismatch (expected 0x{stored_crc:08X}, got 0x{computed:08X})"
+                        );
+                        continue;
+                    }
+
+                    let key = payload[..key_len].to_vec();
+                    let value = payload[key_len..].to_vec();
+                    items.push(WalItem::Record(WalEntry {
+                        seq,
+                        op,
+                        key,
+                        value,
+                    }));
+                }
+                WAL_COMMIT_MAGIC => {
+                    // Read rest: seq(8) + crc(4) = 12
+                    let mut rest = [0u8; 12];
+                    match reader.read_exact(&mut rest) {
+                        Ok(()) => {}
+                        Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => break,
+                        Err(e) => return Err(e.into()),
+                    }
+
+                    let seq = u64::from_le_bytes(rest[0..8].try_into().unwrap());
+                    let stored_crc = u32::from_le_bytes(rest[8..12].try_into().unwrap());
+
+                    let mut hasher = crc32fast::Hasher::new();
+                    hasher.update(&magic_buf);
+                    hasher.update(&rest[0..8]);
+                    let computed = hasher.finalize();
+
+                    if computed != stored_crc {
+                        tracing::warn!(
+                            seq,
+                            "skipping corrupt WAL commit marker: CRC mismatch"
+                        );
+                        continue;
+                    }
+
+                    items.push(WalItem::Commit(seq));
+                }
+                _ => {
+                    // Unknown magic — file is corrupt past this point
+                    tracing::warn!("unknown WAL magic 0x{magic:04X}, stopping scan");
+                    break;
+                }
+            }
+        }
+
+        Ok(items)
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn make_wal(dir: &tempfile::TempDir) -> BinaryWal {
+        let path = dir.path().join("test.wal");
+        let wal = BinaryWal::new(path);
+        wal.initialize().unwrap();
+        wal
+    }
+
+    #[test]
+    fn append_and_commit() {
+        let dir = tempfile::tempdir().unwrap();
+        let wal = make_wal(&dir);
+
+        let seq = wal
+            .append(WalOpType::Insert, b"key1", b"value1")
+            .unwrap();
+        assert_eq!(seq, 1);
+
+        wal.append_commit(seq).unwrap();
+
+        // All committed — recover should return empty
+        let uncommitted = wal.recover().unwrap();
+        assert!(uncommitted.is_empty());
+    }
+
+    #[test]
+    fn uncommitted_entries_recovered() {
+        let dir = tempfile::tempdir().unwrap();
+        let wal = make_wal(&dir);
+
+        let s1 = wal
+            .append(WalOpType::Insert, b"k1", b"v1")
+            .unwrap();
+        wal.append_commit(s1).unwrap();
+
+        // s2 is NOT committed
+        let s2 = wal
+            .append(WalOpType::Update, b"k2", b"v2")
+            .unwrap();
+
+        let uncommitted = wal.recover().unwrap();
+        assert_eq!(uncommitted.len(), 1);
+        assert_eq!(uncommitted[0].seq, s2);
+        assert_eq!(uncommitted[0].op, WalOpType::Update);
+        assert_eq!(uncommitted[0].key, b"k2");
+        assert_eq!(uncommitted[0].value, b"v2");
+    }
+
+    #[test]
+    fn truncate_clears_wal() {
+        let dir = tempfile::tempdir().unwrap();
+        let wal = make_wal(&dir);
+
+        wal.append(WalOpType::Insert, b"k", b"v").unwrap();
+        wal.truncate().unwrap();
+
+        let uncommitted = wal.recover().unwrap();
+        assert!(uncommitted.is_empty());
+    }
+
+    #[test]
+    fn multiple_operations() {
+        let dir = tempfile::tempdir().unwrap();
+        let wal = make_wal(&dir);
+
+        let s1 = wal.append(WalOpType::Insert, b"a", b"1").unwrap();
+        let s2 = wal.append(WalOpType::Update, b"b", b"2").unwrap();
+        let s3 = wal.append(WalOpType::Delete, b"c", b"").unwrap();
+
+        // Commit only s1 and s3
+        wal.append_commit(s1).unwrap();
+        wal.append_commit(s3).unwrap();
+
+        let uncommitted = wal.recover().unwrap();
+        assert_eq!(uncommitted.len(), 1);
+        assert_eq!(uncommitted[0].seq, s2);
+    }
+
+    #[test]
+    fn sequence_numbers_persist_across_reinit() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("persist.wal");
+
+        {
+            let wal = BinaryWal::new(path.clone());
+            wal.initialize().unwrap();
+            let s1 = wal.append(WalOpType::Insert, b"k", b"v").unwrap();
+            assert_eq!(s1, 1);
+            wal.append_commit(s1).unwrap();
+        }
+
+        // Re-open — seq should continue from 2+ (since max committed was 1)
+        {
+            let wal = BinaryWal::new(path);
+            wal.initialize().unwrap();
+            let s2 = wal.append(WalOpType::Insert, b"k2", b"v2").unwrap();
+            assert!(s2 >= 2, "seq should continue: got {s2}");
+        }
+    }
+
+    #[test]
+    fn delete_has_empty_value() {
+        let dir = tempfile::tempdir().unwrap();
+        let wal = make_wal(&dir);
+
+        let seq = wal.append(WalOpType::Delete, b"key", b"").unwrap();
+
+        let uncommitted = wal.recover().unwrap();
+        assert_eq!(uncommitted.len(), 1);
+        assert_eq!(uncommitted[0].seq, seq);
+        assert_eq!(uncommitted[0].op, WalOpType::Delete);
+        assert!(uncommitted[0].value.is_empty());
+    }
+}
@@ -0,0 +1,270 @@
+//! Compaction for the Bitcask-style storage engine.
+//!
+//! Over time, the data file accumulates dead records (superseded by updates,
+//! tombstones from deletes). Compaction rewrites the data file with only live
+//! records, reclaiming disk space.
+//!
+//! The process is:
+//! 1. Create a new `data.rdb.compact` file with a fresh file header.
+//! 2. Iterate all live entries from the KeyDir.
+//! 3. Read each live document from the old data file, write to the new file.
+//! 4. Atomically rename `data.rdb.compact` → `data.rdb`.
+//! 5. Update KeyDir entries with new offsets.
+//! 6. Reset dead_bytes counter.
+
+use std::io::{Seek, SeekFrom, Write};
+use std::path::Path;
+use std::sync::atomic::Ordering;
+
+use tracing::info;
+
+use crate::error::StorageResult;
+use crate::keydir::{KeyDir, KeyDirEntry};
+use crate::record::{DataRecord, FileHeader, FileType, FILE_HEADER_SIZE};
+
+/// Result of a compaction operation.
+#[derive(Debug)]
+pub struct CompactionResult {
+    /// Number of live records written.
+    pub records_written: u64,
+    /// Bytes reclaimed (old file size - new file size).
+    pub bytes_reclaimed: u64,
+    /// New data file size.
+    pub new_file_size: u64,
+}
+
+/// Compact a collection's data file.
+///
+/// This function:
+/// - Reads all live documents (entries present in the KeyDir) from the old data file
+/// - Writes them sequentially to a new file
+/// - Atomically renames the new file over the old one
+/// - Updates all KeyDir entries with their new offsets
+///
+/// The caller must hold the collection's write lock during this operation.
+pub fn compact_data_file(
+    data_path: &Path,
+    keydir: &KeyDir,
+    dead_bytes: &std::sync::atomic::AtomicU64,
+    data_file_size: &std::sync::atomic::AtomicU64,
+) -> StorageResult<CompactionResult> {
+    let compact_path = data_path.with_extension("rdb.compact");
+
+    let old_file_size = std::fs::metadata(data_path)
+        .map(|m| m.len())
+        .unwrap_or(0);
+
+    // Collect all live entries with their keys
+    let mut live_entries: Vec<(String, KeyDirEntry)> = Vec::with_capacity(keydir.len() as usize);
+    keydir.for_each(|key, entry| {
+        live_entries.push((key.to_string(), *entry));
+    });
+
+    // Sort by offset for sequential reads (cache-friendly)
+    live_entries.sort_by_key(|(_, e)| e.offset);
+
+    // Create compact file with header
+    let mut compact_file = std::fs::File::create(&compact_path)?;
+    let hdr = FileHeader::new(FileType::Data);
+    compact_file.write_all(&hdr.encode())?;
+
+    let mut current_offset = FILE_HEADER_SIZE as u64;
+    let mut new_entries: Vec<(String, KeyDirEntry)> = Vec::with_capacity(live_entries.len());
+    let mut old_data_file = std::fs::File::open(data_path)?;
+
+    for (key, entry) in &live_entries {
+        // Read the record from the old file
+        old_data_file.seek(SeekFrom::Start(entry.offset))?;
+        let (record, _disk_size) = DataRecord::decode_from(&mut old_data_file)?
+            .ok_or_else(|| {
+                crate::error::StorageError::CorruptRecord(format!(
+                    "compaction: unexpected EOF reading doc '{key}' at offset {}",
+                    entry.offset
+                ))
+            })?;
+
+        // Write to compact file
+        let encoded = record.encode();
+        let new_disk_size = encoded.len() as u32;
+        compact_file.write_all(&encoded)?;
+
+        new_entries.push((
+            key.clone(),
+            KeyDirEntry {
+                offset: current_offset,
+                record_len: new_disk_size,
+                value_len: entry.value_len,
+                timestamp: entry.timestamp,
+            },
+        ));
+
+        current_offset += new_disk_size as u64;
+    }
+
+    compact_file.sync_all()?;
+    drop(compact_file);
+    drop(old_data_file);
+
+    // Atomic rename
+    std::fs::rename(&compact_path, data_path)?;
+
+    // Update KeyDir with new offsets
+    for (key, new_entry) in new_entries {
+        keydir.insert(key, new_entry);
+    }
+
+    // Reset counters
+    dead_bytes.store(0, Ordering::Relaxed);
+    data_file_size.store(current_offset, Ordering::Relaxed);
+
+    let bytes_reclaimed = old_file_size.saturating_sub(current_offset);
+
+    info!(
+        records = live_entries.len(),
+        old_size = old_file_size,
+        new_size = current_offset,
+        reclaimed = bytes_reclaimed,
+        "compaction complete"
+    );
+
+    Ok(CompactionResult {
+        records_written: live_entries.len() as u64,
+        bytes_reclaimed,
+        new_file_size: current_offset,
+    })
+}
+
+/// Check if compaction is warranted for a collection.
+/// Returns true if dead bytes exceed 50% of live data.
+pub fn should_compact(dead_bytes: u64, data_file_size: u64) -> bool {
+    if data_file_size <= FILE_HEADER_SIZE as u64 {
+        return false;
+    }
+    let useful_bytes = data_file_size - FILE_HEADER_SIZE as u64;
+    // Trigger when dead > 50% of total useful data
+    dead_bytes > useful_bytes / 2
+}
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::keydir::KeyDir;
+    use crate::record::{now_ms, DataRecord, FileHeader, FileType};
+    use std::io::Write;
+    use std::sync::atomic::AtomicU64;
+
+    #[test]
+    fn compact_removes_dead_records() {
+        let dir = tempfile::tempdir().unwrap();
+        let data_path = dir.path().join("data.rdb");
+
+        // Write a data file: insert A, update A (new version), insert B
+        let mut f = std::fs::File::create(&data_path).unwrap();
+        let hdr = FileHeader::new(FileType::Data);
+        f.write_all(&hdr.encode()).unwrap();
+
+        let ts = now_ms();
+
+        // Record 1: A v1 (will be superseded)
+        let r1 = DataRecord {
+            timestamp: ts,
+            key: b"aaa".to_vec(),
+            value: b"old_value".to_vec(),
+        };
+        let r1_enc = r1.encode();
+        let r1_offset = FILE_HEADER_SIZE as u64;
+        let r1_size = r1_enc.len();
+        f.write_all(&r1_enc).unwrap();
+
+        // Record 2: A v2 (current)
+        let r2 = DataRecord {
+            timestamp: ts + 1,
+            key: b"aaa".to_vec(),
+            value: b"new_value".to_vec(),
+        };
+        let r2_enc = r2.encode();
+        let r2_offset = r1_offset + r1_size as u64;
+        let r2_size = r2_enc.len();
+        f.write_all(&r2_enc).unwrap();
+
+        // Record 3: B (live)
+        let r3 = DataRecord {
+            timestamp: ts + 2,
+            key: b"bbb".to_vec(),
+            value: b"bbb_value".to_vec(),
+        };
+        let r3_enc = r3.encode();
+        let r3_offset = r2_offset + r2_size as u64;
+        f.write_all(&r3_enc).unwrap();
+        f.sync_all().unwrap();
+        drop(f);
+
+        let total_size = std::fs::metadata(&data_path).unwrap().len();
+
+        // Build KeyDir — only points to latest versions
+        let keydir = KeyDir::new();
+        keydir.insert(
+            "aaa".into(),
+            KeyDirEntry {
+                offset: r2_offset,
+                record_len: r2_size as u32,
+                value_len: r2.value.len() as u32,
+                timestamp: ts + 1,
+            },
+        );
+        keydir.insert(
+            "bbb".into(),
+            KeyDirEntry {
+                offset: r3_offset,
+                record_len: r3.encode().len() as u32,
+                value_len: r3.value.len() as u32,
+                timestamp: ts + 2,
+            },
+        );
+
+        let dead_bytes_counter = AtomicU64::new(r1_size as u64);
+        let data_file_size_counter = AtomicU64::new(total_size);
+
+        let result = compact_data_file(
+            &data_path,
+            &keydir,
+            &dead_bytes_counter,
+            &data_file_size_counter,
+        )
+        .unwrap();
+
+        assert_eq!(result.records_written, 2);
+        assert!(result.bytes_reclaimed > 0);
+        assert!(result.new_file_size < total_size);
+
+        // Verify dead_bytes was reset
+        assert_eq!(dead_bytes_counter.load(Ordering::Relaxed), 0);
+
+        // Verify KeyDir was updated with new offsets
+        let a_entry = keydir.get("aaa").unwrap();
+        assert_eq!(a_entry.offset, FILE_HEADER_SIZE as u64); // first record after header
+        assert_eq!(a_entry.value_len, b"new_value".len() as u32);
+
+        let b_entry = keydir.get("bbb").unwrap();
+        assert!(b_entry.offset > a_entry.offset);
+
+        // Verify the compacted file can be used to rebuild KeyDir
+        let (rebuilt, dead) = KeyDir::build_from_data_file(&data_path).unwrap();
+        assert_eq!(rebuilt.len(), 2);
+        assert_eq!(dead, 0); // no dead records in compacted file
+    }
+
+    #[test]
+    fn should_compact_thresholds() {
+        // Under threshold
+        assert!(!should_compact(10, 100 + FILE_HEADER_SIZE as u64));
+        // Over threshold (dead > 50% of useful)
+        assert!(should_compact(60, 100 + FILE_HEADER_SIZE as u64));
+        // Empty file
+        assert!(!should_compact(0, FILE_HEADER_SIZE as u64));
+    }
+}
@@ -17,6 +17,15 @@ pub enum StorageError {

    #[error("conflict detected: {0}")]
    ConflictError(String),
+
+    #[error("corrupt record: {0}")]
+    CorruptRecord(String),
+
+    #[error("checksum mismatch: expected 0x{expected:08X}, got 0x{actual:08X}")]
+    ChecksumMismatch { expected: u32, actual: u32 },
+
+    #[error("WAL error: {0}")]
+    WalError(String),
 }

 impl From<serde_json::Error> for StorageError {
@@ -0,0 +1,453 @@
+//! KeyDir — in-memory document location index for the Bitcask storage engine.
+//!
+//! Maps document `_id` (hex string) to its location in the append-only data file.
+//! Backed by `DashMap` for lock-free concurrent reads and fine-grained write locking.
+//!
+//! The KeyDir can be rebuilt from a data file scan, or loaded quickly from a
+//! persisted hint file for fast restart.
+
+use std::io::{self, BufReader, BufWriter, Read, Write};
+use std::path::Path;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use dashmap::DashMap;
+
+use crate::error::{StorageError, StorageResult};
+use crate::record::{
+    FileHeader, FileType, RecordScanner, FILE_HEADER_SIZE, FORMAT_VERSION,
+};
+
+// ---------------------------------------------------------------------------
+// KeyDirEntry
+// ---------------------------------------------------------------------------
+
+/// Location of a single document in the data file.
+#[derive(Debug, Clone, Copy)]
+pub struct KeyDirEntry {
+    /// Byte offset of the record in `data.rdb`.
+    pub offset: u64,
+    /// Total record size on disk (header + payload).
+    pub record_len: u32,
+    /// BSON value length. 0 means tombstone (used during compaction accounting).
+    pub value_len: u32,
+    /// Timestamp (epoch ms) from the record. Used for conflict detection.
+    pub timestamp: u64,
+}
+
+// ---------------------------------------------------------------------------
+// KeyDir
+// ---------------------------------------------------------------------------
+
+/// In-memory index mapping document ID → data file location.
+pub struct KeyDir {
+    map: DashMap<String, KeyDirEntry>,
+    /// Running count of live documents.
+    doc_count: AtomicU64,
+}
+
+impl KeyDir {
+    /// Create an empty KeyDir.
+    pub fn new() -> Self {
+        Self {
+            map: DashMap::new(),
+            doc_count: AtomicU64::new(0),
+        }
+    }
+
+    /// Insert or update an entry. Returns the previous entry if one existed.
+    pub fn insert(&self, key: String, entry: KeyDirEntry) -> Option<KeyDirEntry> {
+        let prev = self.map.insert(key, entry);
+        if prev.is_none() {
+            self.doc_count.fetch_add(1, Ordering::Relaxed);
+        }
+        prev
+    }
+
+    /// Look up an entry by key.
+    pub fn get(&self, key: &str) -> Option<KeyDirEntry> {
+        self.map.get(key).map(|r| *r.value())
+    }
+
+    /// Remove an entry. Returns the removed entry if it existed.
+    pub fn remove(&self, key: &str) -> Option<KeyDirEntry> {
+        let removed = self.map.remove(key).map(|(_, v)| v);
+        if removed.is_some() {
+            self.doc_count.fetch_sub(1, Ordering::Relaxed);
+        }
+        removed
+    }
+
+    /// Number of live documents.
+    pub fn len(&self) -> u64 {
+        self.doc_count.load(Ordering::Relaxed)
+    }
+
+    /// Whether the index is empty.
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// Check if a key exists.
+    pub fn contains(&self, key: &str) -> bool {
+        self.map.contains_key(key)
+    }
+
+    /// Iterate over all entries. The closure receives (key, entry).
+    pub fn for_each(&self, mut f: impl FnMut(&str, &KeyDirEntry)) {
+        for entry in self.map.iter() {
+            f(entry.key(), entry.value());
+        }
+    }
+
+    /// Collect all keys.
+    pub fn keys(&self) -> Vec<String> {
+        self.map.iter().map(|e| e.key().clone()).collect()
+    }
+
+    /// Clear all entries.
+    pub fn clear(&self) {
+        self.map.clear();
+        self.doc_count.store(0, Ordering::Relaxed);
+    }
+
+    // -----------------------------------------------------------------------
+    // Build from data file
+    // -----------------------------------------------------------------------
+
+    /// Rebuild the KeyDir by scanning an entire data file.
+    /// The file must start with a valid `FileHeader`.
+    /// Returns `(keydir, dead_bytes)` where `dead_bytes` is the total size of
+    /// stale records (superseded by later writes or tombstoned).
+    pub fn build_from_data_file(path: &Path) -> StorageResult<(Self, u64)> {
+        let file = std::fs::File::open(path)?;
+        let mut reader = BufReader::new(file);
+
+        // Read and validate file header
+        let mut hdr_buf = [0u8; FILE_HEADER_SIZE];
+        reader.read_exact(&mut hdr_buf)?;
+        let hdr = FileHeader::decode(&hdr_buf)?;
+        if hdr.file_type != FileType::Data {
+            return Err(StorageError::CorruptRecord(format!(
+                "expected data file (type 1), got type {:?}",
+                hdr.file_type
+            )));
+        }
+
+        let keydir = KeyDir::new();
+        let mut dead_bytes: u64 = 0;
+
+        let scanner = RecordScanner::new(reader, FILE_HEADER_SIZE as u64);
+        for result in scanner {
+            let (offset, record) = result?;
+            let is_tombstone = record.is_tombstone();
+            let disk_size = record.disk_size() as u32;
+            let value_len = record.value.len() as u32;
+            let timestamp = record.timestamp;
+            let key = String::from_utf8(record.key)
+                .map_err(|e| StorageError::CorruptRecord(format!("invalid UTF-8 key: {e}")))?;
+
+            if is_tombstone {
+                // Remove from index; the tombstone itself is dead weight
+                if let Some(prev) = keydir.remove(&key) {
+                    dead_bytes += prev.record_len as u64;
+                }
+                dead_bytes += disk_size as u64;
+            } else {
+                let entry = KeyDirEntry {
+                    offset,
+                    record_len: disk_size,
+                    value_len,
+                    timestamp,
+                };
+                if let Some(prev) = keydir.insert(key, entry) {
+                    // Previous version of same key is now dead
+                    dead_bytes += prev.record_len as u64;
+                }
+            }
+        }
+
+        Ok((keydir, dead_bytes))
+    }
+
+    // -----------------------------------------------------------------------
+    // Hint file persistence (for fast startup)
+    // -----------------------------------------------------------------------
+
+    /// Persist the KeyDir to a hint file for fast restart.
+    ///
+    /// Hint file format (after the 64-byte file header):
+    /// For each entry: [key_len:u32 LE][key bytes][offset:u64 LE][record_len:u32 LE][value_len:u32 LE][timestamp:u64 LE]
+    pub fn persist_to_hint_file(&self, path: &Path) -> StorageResult<()> {
+        let file = std::fs::File::create(path)?;
+        let mut writer = BufWriter::new(file);
+
+        // Write file header
+        let hdr = FileHeader::new(FileType::Hint);
+        writer.write_all(&hdr.encode())?;
+
+        // Write entries
+        for entry in self.map.iter() {
+            let key_bytes = entry.key().as_bytes();
+            let key_len = key_bytes.len() as u32;
+            writer.write_all(&key_len.to_le_bytes())?;
+            writer.write_all(key_bytes)?;
+            writer.write_all(&entry.value().offset.to_le_bytes())?;
+            writer.write_all(&entry.value().record_len.to_le_bytes())?;
+            writer.write_all(&entry.value().value_len.to_le_bytes())?;
+            writer.write_all(&entry.value().timestamp.to_le_bytes())?;
+        }
+
+        writer.flush()?;
+        Ok(())
+    }
+
+    /// Load a KeyDir from a hint file. Returns None if the file doesn't exist.
+    pub fn load_from_hint_file(path: &Path) -> StorageResult<Option<Self>> {
+        if !path.exists() {
+            return Ok(None);
+        }
+
+        let file = std::fs::File::open(path)?;
+        let mut reader = BufReader::new(file);
+
+        // Read and validate header
+        let mut hdr_buf = [0u8; FILE_HEADER_SIZE];
+        match reader.read_exact(&mut hdr_buf) {
+            Ok(()) => {}
+            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),
+            Err(e) => return Err(e.into()),
+        }
+        let hdr = FileHeader::decode(&hdr_buf)?;
+        if hdr.file_type != FileType::Hint {
+            return Err(StorageError::CorruptRecord(format!(
+                "expected hint file (type 3), got type {:?}",
+                hdr.file_type
+            )));
+        }
+        if hdr.version > FORMAT_VERSION {
+            return Err(StorageError::CorruptRecord(format!(
+                "hint file version {} is newer than supported {}",
+                hdr.version, FORMAT_VERSION
+            )));
+        }
+
+        let keydir = KeyDir::new();
+
+        loop {
+            // Read key_len
+            let mut key_len_buf = [0u8; 4];
+            match reader.read_exact(&mut key_len_buf) {
+                Ok(()) => {}
+                Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => break,
+                Err(e) => return Err(e.into()),
+            }
+            let key_len = u32::from_le_bytes(key_len_buf) as usize;
+
+            // Read key
+            let mut key_buf = vec![0u8; key_len];
+            reader.read_exact(&mut key_buf)?;
+            let key = String::from_utf8(key_buf)
+                .map_err(|e| StorageError::CorruptRecord(format!("invalid UTF-8 key: {e}")))?;
+
+            // Read entry fields
+            let mut fields = [0u8; 8 + 4 + 4 + 8]; // offset + record_len + value_len + timestamp = 24
+            reader.read_exact(&mut fields)?;
+
+            let offset = u64::from_le_bytes(fields[0..8].try_into().unwrap());
+            let record_len = u32::from_le_bytes(fields[8..12].try_into().unwrap());
+            let value_len = u32::from_le_bytes(fields[12..16].try_into().unwrap());
+            let timestamp = u64::from_le_bytes(fields[16..24].try_into().unwrap());
+
+            keydir.insert(
+                key,
+                KeyDirEntry {
+                    offset,
+                    record_len,
+                    value_len,
+                    timestamp,
+                },
+            );
+        }
+
+        Ok(Some(keydir))
+    }
+}
+
+impl Default for KeyDir {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::record::DataRecord;
+    use std::io::Write;
+
+    #[test]
+    fn basic_insert_get_remove() {
+        let kd = KeyDir::new();
+        assert!(kd.is_empty());
+
+        let entry = KeyDirEntry {
+            offset: 100,
+            record_len: 50,
+            value_len: 30,
+            timestamp: 1700000000000,
+        };
+
+        assert!(kd.insert("abc".into(), entry).is_none());
+        assert_eq!(kd.len(), 1);
+        assert!(kd.contains("abc"));
+
+        let got = kd.get("abc").unwrap();
+        assert_eq!(got.offset, 100);
+        assert_eq!(got.value_len, 30);
+
+        let removed = kd.remove("abc").unwrap();
+        assert_eq!(removed.offset, 100);
+        assert_eq!(kd.len(), 0);
+        assert!(!kd.contains("abc"));
+    }
+
+    #[test]
+    fn insert_overwrites_returns_previous() {
+        let kd = KeyDir::new();
+        let e1 = KeyDirEntry {
+            offset: 100,
+            record_len: 50,
+            value_len: 30,
+            timestamp: 1,
+        };
+        let e2 = KeyDirEntry {
+            offset: 200,
+            record_len: 60,
+            value_len: 40,
+            timestamp: 2,
+        };
+
+        kd.insert("k".into(), e1);
+        assert_eq!(kd.len(), 1);
+
+        let prev = kd.insert("k".into(), e2).unwrap();
+        assert_eq!(prev.offset, 100);
+        // Count stays at 1 (overwrite, not new)
+        assert_eq!(kd.len(), 1);
+        assert_eq!(kd.get("k").unwrap().offset, 200);
+    }
+
+    #[test]
+    fn build_from_data_file() {
+        let dir = tempfile::tempdir().unwrap();
+        let data_path = dir.path().join("data.rdb");
+
+        // Write a data file with 3 records: insert A, insert B, delete A
+        {
+            let mut f = std::fs::File::create(&data_path).unwrap();
+            let hdr = FileHeader::new(FileType::Data);
+            f.write_all(&hdr.encode()).unwrap();
+
+            let r1 = DataRecord {
+                timestamp: 1,
+                key: b"aaa".to_vec(),
+                value: b"val_a".to_vec(),
+            };
+            let r2 = DataRecord {
+                timestamp: 2,
+                key: b"bbb".to_vec(),
+                value: b"val_b".to_vec(),
+            };
+            let r3 = DataRecord {
+                timestamp: 3,
+                key: b"aaa".to_vec(),
+                value: vec![], // tombstone
+            };
+            f.write_all(&r1.encode()).unwrap();
+            f.write_all(&r2.encode()).unwrap();
+            f.write_all(&r3.encode()).unwrap();
+        }
+
+        let (kd, dead_bytes) = KeyDir::build_from_data_file(&data_path).unwrap();
+
+        // Only B should be live
+        assert_eq!(kd.len(), 1);
+        assert!(kd.contains("bbb"));
+        assert!(!kd.contains("aaa"));
+
+        // Dead bytes: r1 (aaa live, then superseded by tombstone) + r3 (tombstone itself)
+        assert!(dead_bytes > 0);
+    }
+
+    #[test]
+    fn hint_file_roundtrip() {
+        let dir = tempfile::tempdir().unwrap();
+        let hint_path = dir.path().join("keydir.hint");
+
+        let kd = KeyDir::new();
+        kd.insert(
+            "doc1".into(),
+            KeyDirEntry {
+                offset: 64,
+                record_len: 100,
+                value_len: 80,
+                timestamp: 1000,
+            },
+        );
+        kd.insert(
+            "doc2".into(),
+            KeyDirEntry {
+                offset: 164,
+                record_len: 200,
+                value_len: 150,
+                timestamp: 2000,
+            },
+        );
+
+        kd.persist_to_hint_file(&hint_path).unwrap();
+        let loaded = KeyDir::load_from_hint_file(&hint_path).unwrap().unwrap();
+
+        assert_eq!(loaded.len(), 2);
+        let e1 = loaded.get("doc1").unwrap();
+        assert_eq!(e1.offset, 64);
+        assert_eq!(e1.record_len, 100);
+        assert_eq!(e1.value_len, 80);
+        assert_eq!(e1.timestamp, 1000);
+
+        let e2 = loaded.get("doc2").unwrap();
+        assert_eq!(e2.offset, 164);
+        assert_eq!(e2.timestamp, 2000);
+    }
+
+    #[test]
+    fn hint_file_nonexistent_returns_none() {
+        let result = KeyDir::load_from_hint_file(Path::new("/tmp/nonexistent_hint_file.hint"));
+        assert!(result.unwrap().is_none());
+    }
+
+    #[test]
+    fn for_each_and_keys() {
+        let kd = KeyDir::new();
+        let e = KeyDirEntry {
+            offset: 0,
+            record_len: 10,
+            value_len: 5,
+            timestamp: 1,
+        };
+        kd.insert("x".into(), e);
+        kd.insert("y".into(), e);
+
+        let mut collected = Vec::new();
+        kd.for_each(|k, _| collected.push(k.to_string()));
+        collected.sort();
+        assert_eq!(collected, vec!["x", "y"]);
+
+        let mut keys = kd.keys();
+        keys.sort();
+        assert_eq!(keys, vec!["x", "y"]);
+    }
+}
@@ -2,21 +2,30 @@
 //!
 //! Provides the [`StorageAdapter`] trait and two concrete implementations:
 //! - [`MemoryStorageAdapter`] -- fast in-memory store backed by `DashMap`
-//! - [`FileStorageAdapter`] -- JSON-file-per-collection persistent store
+//! - [`FileStorageAdapter`] -- Bitcask-style append-only log with crash recovery
 //!
-//! Also includes an [`OpLog`] for operation logging and a [`WriteAheadLog`]
-//! for crash recovery.
+//! Also includes an [`OpLog`] for operation logging, a [`BinaryWal`] for
+//! write-ahead logging, and [`compaction`] for dead record reclamation.

 pub mod adapter;
+pub mod binary_wal;
+pub mod compaction;
 pub mod error;
 pub mod file;
+pub mod keydir;
 pub mod memory;
 pub mod oplog;
-pub mod wal;
+pub mod record;

 pub use adapter::StorageAdapter;
+pub use binary_wal::{BinaryWal, WalEntry, WalOpType};
+pub use compaction::{compact_data_file, should_compact, CompactionResult};
 pub use error::{StorageError, StorageResult};
 pub use file::FileStorageAdapter;
+pub use keydir::{KeyDir, KeyDirEntry};
 pub use memory::MemoryStorageAdapter;
-pub use oplog::{OpLog, OpLogEntry, OpType};
-pub use wal::{WalOp, WalRecord, WriteAheadLog};
+pub use oplog::{OpLog, OpLogEntry, OpLogStats, OpType};
+pub use record::{
+    DataRecord, FileHeader, FileType, RecordScanner, FILE_HEADER_SIZE, FILE_MAGIC, FORMAT_VERSION,
+    RECORD_HEADER_SIZE, RECORD_MAGIC,
+};
@@ -2,6 +2,8 @@
 //!
 //! The OpLog records every write operation so that changes can be replayed,
 //! replicated, or used for change-stream style notifications.
+//! Each entry stores both the new and previous document state, enabling
+//! point-in-time replay and revert.

 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{SystemTime, UNIX_EPOCH};
@@ -33,8 +35,21 @@ pub struct OpLogEntry {
    pub collection: String,
    /// Document id (hex string).
    pub document_id: String,
-    /// The document snapshot (for insert/update; None for delete).
+    /// The new document snapshot (for insert/update; None for delete).
    pub document: Option<Document>,
+    /// The previous document snapshot (for update/delete; None for insert).
+    pub previous_document: Option<Document>,
+}
+
+/// Aggregate statistics about the oplog.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct OpLogStats {
+    pub current_seq: u64,
+    pub total_entries: usize,
+    pub oldest_seq: u64,
+    pub inserts: usize,
+    pub updates: usize,
+    pub deletes: usize,
 }

 /// In-memory operation log.
@@ -61,6 +76,7 @@ impl OpLog {
        collection: &str,
        document_id: &str,
        document: Option<Document>,
+        previous_document: Option<Document>,
    ) -> u64 {
        let seq = self.next_seq.fetch_add(1, Ordering::SeqCst);
        let entry = OpLogEntry {
@@ -74,11 +90,17 @@ impl OpLog {
            collection: collection.to_string(),
            document_id: document_id.to_string(),
            document,
+            previous_document,
        };
        self.entries.insert(seq, entry);
        seq
    }

+    /// Get a single entry by sequence number.
+    pub fn get_entry(&self, seq: u64) -> Option<OpLogEntry> {
+        self.entries.get(&seq).map(|e| e.value().clone())
+    }
+
    /// Get all entries with sequence number >= `since`.
    pub fn entries_since(&self, since: u64) -> Vec<OpLogEntry> {
        let mut result: Vec<_> = self
@@ -91,11 +113,72 @@ impl OpLog {
        result
    }

+    /// Get entries in range [from_seq, to_seq] inclusive, sorted by seq.
+    pub fn entries_range(&self, from_seq: u64, to_seq: u64) -> Vec<OpLogEntry> {
+        let mut result: Vec<_> = self
+            .entries
+            .iter()
+            .filter(|e| {
+                let k = *e.key();
+                k >= from_seq && k <= to_seq
+            })
+            .map(|e| e.value().clone())
+            .collect();
+        result.sort_by_key(|e| e.seq);
+        result
+    }
+
+    /// Remove all entries with seq > `after_seq` and reset the next_seq counter.
+    pub fn truncate_after(&self, after_seq: u64) {
+        let keys_to_remove: Vec<u64> = self
+            .entries
+            .iter()
+            .filter(|e| *e.key() > after_seq)
+            .map(|e| *e.key())
+            .collect();
+        for key in keys_to_remove {
+            self.entries.remove(&key);
+        }
+        self.next_seq.store(after_seq + 1, Ordering::SeqCst);
+    }
+
    /// Get the current (latest) sequence number. Returns 0 if empty.
    pub fn current_seq(&self) -> u64 {
        self.next_seq.load(Ordering::SeqCst).saturating_sub(1)
    }

+    /// Get aggregate statistics.
+    pub fn stats(&self) -> OpLogStats {
+        let mut inserts = 0usize;
+        let mut updates = 0usize;
+        let mut deletes = 0usize;
+        let mut oldest_seq = u64::MAX;
+
+        for entry in self.entries.iter() {
+            match entry.value().op {
+                OpType::Insert => inserts += 1,
+                OpType::Update => updates += 1,
+                OpType::Delete => deletes += 1,
+            }
+            if entry.value().seq < oldest_seq {
+                oldest_seq = entry.value().seq;
+            }
+        }
+
+        if oldest_seq == u64::MAX {
+            oldest_seq = 0;
+        }
+
+        OpLogStats {
+            current_seq: self.current_seq(),
+            total_entries: self.entries.len(),
+            oldest_seq,
+            inserts,
+            updates,
+            deletes,
+        }
+    }
+
    /// Clear all entries.
    pub fn clear(&self) {
        self.entries.clear();
@@ -0,0 +1,452 @@
+//! Binary data record format for the Bitcask-style storage engine.
+//!
+//! # File Version Header (64 bytes, at offset 0 of every .rdb / .hint file)
+//!
+//! ```text
+//! ┌──────────────┬──────────┬──────────┬──────────┬──────────┬───────────────┐
+//! │ magic        │ version  │ file_type│ flags    │ created  │ reserved      │
+//! │ 8 bytes      │ u16 LE   │ u8       │ u32 LE   │ u64 LE   │ 41 bytes      │
+//! │ "SMARTDB\0"  │          │          │          │ epoch_ms │ (zeros)       │
+//! └──────────────┴──────────┴──────────┴──────────┴──────────┴───────────────┘
+//! ```
+//!
+//! # Data Record (appended after the header)
+//!
+//! ```text
+//! ┌──────────┬──────────┬──────────┬──────────┬──────────┬──────────────────┐
+//! │ magic    │ timestamp│ key_len  │ val_len  │ crc32    │ payload          │
+//! │ u16 LE   │ u64 LE   │ u32 LE   │ u32 LE   │ u32 LE   │ [key][value]     │
+//! │ 0xDB01   │ epoch_ms │          │ 0=delete │          │                  │
+//! └──────────┴──────────┴──────────┴──────────┴──────────┴──────────────────┘
+//! ```
+
+use std::io::{self, Read};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use crate::error::{StorageError, StorageResult};
+
+// ---------------------------------------------------------------------------
+// Constants
+// ---------------------------------------------------------------------------
+
+/// File-level magic: b"SMARTDB\0"
+pub const FILE_MAGIC: &[u8; 8] = b"SMARTDB\0";
+
+/// Current storage format version.
+pub const FORMAT_VERSION: u16 = 1;
+
+/// File version header size.
+pub const FILE_HEADER_SIZE: usize = 64;
+
+/// Per-record magic.
+pub const RECORD_MAGIC: u16 = 0xDB01;
+
+/// Per-record header size (before payload).
+pub const RECORD_HEADER_SIZE: usize = 2 + 8 + 4 + 4 + 4; // 22 bytes
+
+// ---------------------------------------------------------------------------
+// File type tag stored in the version header
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum FileType {
+    Data = 1,
+    Wal = 2,
+    Hint = 3,
+}
+
+impl FileType {
+    pub fn from_u8(v: u8) -> StorageResult<Self> {
+        match v {
+            1 => Ok(FileType::Data),
+            2 => Ok(FileType::Wal),
+            3 => Ok(FileType::Hint),
+            _ => Err(StorageError::CorruptRecord(format!(
+                "unknown file type tag: {v}"
+            ))),
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// File Version Header
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone)]
+pub struct FileHeader {
+    pub version: u16,
+    pub file_type: FileType,
+    pub flags: u32,
+    pub created_ms: u64,
+}
+
+impl FileHeader {
+    /// Create a new header for the current format version.
+    pub fn new(file_type: FileType) -> Self {
+        Self {
+            version: FORMAT_VERSION,
+            file_type,
+            flags: 0,
+            created_ms: now_ms(),
+        }
+    }
+
+    /// Encode the header to a 64-byte buffer.
+    pub fn encode(&self) -> [u8; FILE_HEADER_SIZE] {
+        let mut buf = [0u8; FILE_HEADER_SIZE];
+        buf[0..8].copy_from_slice(FILE_MAGIC);
+        buf[8..10].copy_from_slice(&self.version.to_le_bytes());
+        buf[10] = self.file_type as u8;
+        buf[11..15].copy_from_slice(&self.flags.to_le_bytes());
+        buf[15..23].copy_from_slice(&self.created_ms.to_le_bytes());
+        // bytes 23..64 are reserved (zeros)
+        buf
+    }
+
+    /// Decode a 64-byte header. Validates magic and version.
+    pub fn decode(buf: &[u8; FILE_HEADER_SIZE]) -> StorageResult<Self> {
+        if &buf[0..8] != FILE_MAGIC {
+            return Err(StorageError::CorruptRecord(
+                "invalid file magic — not a SmartDB file".into(),
+            ));
+        }
+        let version = u16::from_le_bytes([buf[8], buf[9]]);
+        if version > FORMAT_VERSION {
+            return Err(StorageError::CorruptRecord(format!(
+                "file format version {version} is newer than supported version {FORMAT_VERSION} — please upgrade"
+            )));
+        }
+        if version == 0 {
+            return Err(StorageError::CorruptRecord(
+                "file format version 0 is invalid".into(),
+            ));
+        }
+        let file_type = FileType::from_u8(buf[10])?;
+        let flags = u32::from_le_bytes([buf[11], buf[12], buf[13], buf[14]]);
+        let created_ms = u64::from_le_bytes([
+            buf[15], buf[16], buf[17], buf[18], buf[19], buf[20], buf[21], buf[22],
+        ]);
+        Ok(Self {
+            version,
+            file_type,
+            flags,
+            created_ms,
+        })
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Data Record
+// ---------------------------------------------------------------------------
+
+/// A single data record (live document or tombstone).
+#[derive(Debug, Clone)]
+pub struct DataRecord {
+    pub timestamp: u64,
+    pub key: Vec<u8>,
+    /// BSON value bytes. Empty for tombstones.
+    pub value: Vec<u8>,
+}
+
+impl DataRecord {
+    /// Whether this record is a tombstone (delete marker).
+    pub fn is_tombstone(&self) -> bool {
+        self.value.is_empty()
+    }
+
+    /// Total size on disk (header + payload).
+    pub fn disk_size(&self) -> usize {
+        RECORD_HEADER_SIZE + self.key.len() + self.value.len()
+    }
+
+    /// Encode to bytes. CRC32 covers magic + timestamp + key_len + val_len + payload.
+    pub fn encode(&self) -> Vec<u8> {
+        let key_len = self.key.len() as u32;
+        let val_len = self.value.len() as u32;
+        let total = RECORD_HEADER_SIZE + self.key.len() + self.value.len();
+        let mut buf = Vec::with_capacity(total);
+
+        // Write fields WITHOUT crc first to compute checksum.
+        buf.extend_from_slice(&RECORD_MAGIC.to_le_bytes()); // 2
+        buf.extend_from_slice(&self.timestamp.to_le_bytes()); // 8
+        buf.extend_from_slice(&key_len.to_le_bytes()); // 4
+        buf.extend_from_slice(&val_len.to_le_bytes()); // 4
+        // placeholder for crc32 — we'll fill it after computing
+        buf.extend_from_slice(&0u32.to_le_bytes()); // 4
+        buf.extend_from_slice(&self.key); // key_len
+        buf.extend_from_slice(&self.value); // val_len
+
+        // CRC covers everything except the crc32 field itself:
+        // bytes [0..18] (magic+ts+key_len+val_len) + bytes [22..] (payload)
+        let mut hasher = crc32fast::Hasher::new();
+        hasher.update(&buf[0..18]);
+        hasher.update(&buf[22..]);
+        let crc = hasher.finalize();
+        buf[18..22].copy_from_slice(&crc.to_le_bytes());
+
+        buf
+    }
+
+    /// Decode a record from a reader. Returns the record and its total disk size.
+    /// On EOF at the very start (no bytes to read), returns Ok(None).
+    pub fn decode_from<R: Read>(reader: &mut R) -> StorageResult<Option<(Self, usize)>> {
+        // Read header
+        let mut hdr = [0u8; RECORD_HEADER_SIZE];
+        match reader.read_exact(&mut hdr) {
+            Ok(()) => {}
+            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),
+            Err(e) => return Err(e.into()),
+        }
+
+        let magic = u16::from_le_bytes([hdr[0], hdr[1]]);
+        if magic != RECORD_MAGIC {
+            return Err(StorageError::CorruptRecord(format!(
+                "invalid record magic: 0x{magic:04X}, expected 0x{RECORD_MAGIC:04X}"
+            )));
+        }
+
+        let timestamp = u64::from_le_bytes(hdr[2..10].try_into().unwrap());
+        let key_len = u32::from_le_bytes(hdr[10..14].try_into().unwrap()) as usize;
+        let val_len = u32::from_le_bytes(hdr[14..18].try_into().unwrap()) as usize;
+        let stored_crc = u32::from_le_bytes(hdr[18..22].try_into().unwrap());
+
+        // Read payload
+        let payload_len = key_len + val_len;
+        let mut payload = vec![0u8; payload_len];
+        reader.read_exact(&mut payload)?;
+
+        // Verify CRC: covers header bytes [0..18] + payload
+        let mut hasher = crc32fast::Hasher::new();
+        hasher.update(&hdr[0..18]);
+        hasher.update(&payload);
+        let computed_crc = hasher.finalize();
+        if computed_crc != stored_crc {
+            return Err(StorageError::ChecksumMismatch {
+                expected: stored_crc,
+                actual: computed_crc,
+            });
+        }
+
+        let key = payload[..key_len].to_vec();
+        let value = payload[key_len..].to_vec();
+        let disk_size = RECORD_HEADER_SIZE + payload_len;
+
+        Ok(Some((
+            DataRecord {
+                timestamp,
+                key,
+                value,
+            },
+            disk_size,
+        )))
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Record Scanner — iterate records from a byte slice or reader
+// ---------------------------------------------------------------------------
+
+/// Scans records sequentially from a reader, yielding (offset, record) pairs.
+/// Starts reading from the current reader position. The `base_offset` parameter
+/// indicates the byte offset in the file where reading begins (typically
+/// `FILE_HEADER_SIZE` for a data file).
+pub struct RecordScanner<R> {
+    reader: R,
+    offset: u64,
+}
+
+impl<R: Read> RecordScanner<R> {
+    pub fn new(reader: R, base_offset: u64) -> Self {
+        Self {
+            reader,
+            offset: base_offset,
+        }
+    }
+}
+
+impl<R: Read> Iterator for RecordScanner<R> {
+    /// (file_offset, record) or an error. Iteration stops on EOF or error.
+    type Item = StorageResult<(u64, DataRecord)>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match DataRecord::decode_from(&mut self.reader) {
+            Ok(Some((record, disk_size))) => {
+                let offset = self.offset;
+                self.offset += disk_size as u64;
+                Some(Ok((offset, record)))
+            }
+            Ok(None) => None, // clean EOF
+            Err(e) => Some(Err(e)),
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+/// Current time in milliseconds since UNIX epoch.
+pub fn now_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_millis() as u64
+}
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn file_header_roundtrip() {
+        let hdr = FileHeader::new(FileType::Data);
+        let buf = hdr.encode();
+        assert_eq!(buf.len(), FILE_HEADER_SIZE);
+
+        let decoded = FileHeader::decode(&buf).unwrap();
+        assert_eq!(decoded.version, FORMAT_VERSION);
+        assert_eq!(decoded.file_type, FileType::Data);
+        assert_eq!(decoded.flags, 0);
+        assert_eq!(decoded.created_ms, hdr.created_ms);
+    }
+
+    #[test]
+    fn file_header_rejects_bad_magic() {
+        let mut buf = [0u8; FILE_HEADER_SIZE];
+        buf[0..8].copy_from_slice(b"BADMAGIC");
+        assert!(FileHeader::decode(&buf).is_err());
+    }
+
+    #[test]
+    fn file_header_rejects_future_version() {
+        let mut hdr = FileHeader::new(FileType::Data);
+        hdr.version = FORMAT_VERSION + 1;
+        let buf = hdr.encode();
+        // Manually patch the version in the buffer
+        let mut buf2 = buf;
+        buf2[8..10].copy_from_slice(&(FORMAT_VERSION + 1).to_le_bytes());
+        assert!(FileHeader::decode(&buf2).is_err());
+    }
+
+    #[test]
+    fn record_roundtrip_live() {
+        let rec = DataRecord {
+            timestamp: 1700000000000,
+            key: b"abc123".to_vec(),
+            value: b"\x10\x00\x00\x00\x02hi\x00\x03\x00\x00\x00ok\x00\x00".to_vec(),
+        };
+        let encoded = rec.encode();
+        assert_eq!(encoded.len(), rec.disk_size());
+
+        let mut cursor = std::io::Cursor::new(&encoded);
+        let (decoded, size) = DataRecord::decode_from(&mut cursor).unwrap().unwrap();
+        assert_eq!(size, encoded.len());
+        assert_eq!(decoded.timestamp, rec.timestamp);
+        assert_eq!(decoded.key, rec.key);
+        assert_eq!(decoded.value, rec.value);
+        assert!(!decoded.is_tombstone());
+    }
+
+    #[test]
+    fn record_roundtrip_tombstone() {
+        let rec = DataRecord {
+            timestamp: 1700000000000,
+            key: b"def456".to_vec(),
+            value: vec![],
+        };
+        assert!(rec.is_tombstone());
+        let encoded = rec.encode();
+
+        let mut cursor = std::io::Cursor::new(&encoded);
+        let (decoded, _) = DataRecord::decode_from(&mut cursor).unwrap().unwrap();
+        assert!(decoded.is_tombstone());
+        assert_eq!(decoded.key, b"def456");
+    }
+
+    #[test]
+    fn record_detects_corruption() {
+        let rec = DataRecord {
+            timestamp: 42,
+            key: b"key".to_vec(),
+            value: b"value".to_vec(),
+        };
+        let mut encoded = rec.encode();
+        // Flip a bit in the payload
+        let last = encoded.len() - 1;
+        encoded[last] ^= 0xFF;
+
+        let mut cursor = std::io::Cursor::new(&encoded);
+        let result = DataRecord::decode_from(&mut cursor);
+        assert!(matches!(result, Err(StorageError::ChecksumMismatch { .. })));
+    }
+
+    #[test]
+    fn record_detects_bad_magic() {
+        let rec = DataRecord {
+            timestamp: 42,
+            key: b"key".to_vec(),
+            value: b"value".to_vec(),
+        };
+        let mut encoded = rec.encode();
+        encoded[0] = 0xFF;
+        encoded[1] = 0xFF;
+
+        let mut cursor = std::io::Cursor::new(&encoded);
+        let result = DataRecord::decode_from(&mut cursor);
+        assert!(matches!(result, Err(StorageError::CorruptRecord(_))));
+    }
+
+    #[test]
+    fn eof_returns_none() {
+        let empty: &[u8] = &[];
+        let mut cursor = std::io::Cursor::new(empty);
+        let result = DataRecord::decode_from(&mut cursor).unwrap();
+        assert!(result.is_none());
+    }
+
+    #[test]
+    fn scanner_iterates_multiple_records() {
+        let records = vec![
+            DataRecord {
+                timestamp: 1,
+                key: b"a".to_vec(),
+                value: b"v1".to_vec(),
+            },
+            DataRecord {
+                timestamp: 2,
+                key: b"b".to_vec(),
+                value: b"v2".to_vec(),
+            },
+            DataRecord {
+                timestamp: 3,
+                key: b"c".to_vec(),
+                value: vec![],
+            },
+        ];
+
+        let mut buf = Vec::new();
+        for r in &records {
+            buf.extend_from_slice(&r.encode());
+        }
+
+        let scanner = RecordScanner::new(std::io::Cursor::new(&buf), 0);
+        let results: Vec<_> = scanner.collect::<Result<Vec<_>, _>>().unwrap();
+        assert_eq!(results.len(), 3);
+        assert_eq!(results[0].1.key, b"a");
+        assert_eq!(results[1].1.key, b"b");
+        assert!(results[2].1.is_tombstone());
+
+        // Verify offsets are correct
+        assert_eq!(results[0].0, 0);
+        assert_eq!(results[1].0, records[0].disk_size() as u64);
+        assert_eq!(
+            results[2].0,
+            (records[0].disk_size() + records[1].disk_size()) as u64
+        );
+    }
+}
@@ -1,186 +0,0 @@
-//! Write-Ahead Log (WAL) for crash recovery.
-//!
-//! Before any mutation is applied to storage, it is first written to the WAL.
-//! On recovery, uncommitted WAL entries can be replayed or discarded.
-
-use std::path::PathBuf;
-use std::sync::atomic::{AtomicU64, Ordering};
-
-use bson::Document;
-use serde::{Deserialize, Serialize};
-use tokio::io::AsyncWriteExt;
-use tracing::{debug, warn};
-
-use crate::error::StorageResult;
-
-/// WAL operation kind.
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
-pub enum WalOp {
-    Insert,
-    Update,
-    Delete,
-}
-
-/// A single WAL record.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct WalRecord {
-    /// Sequence number.
-    pub seq: u64,
-    /// Operation kind.
-    pub op: WalOp,
-    /// Database name.
-    pub db: String,
-    /// Collection name.
-    pub collection: String,
-    /// Document id (hex string).
-    pub document_id: String,
-    /// Document data (for insert/update).
-    pub document: Option<Document>,
-    /// Whether this record has been committed (applied to storage).
-    pub committed: bool,
-    /// CRC32 checksum of the serialized payload for integrity verification.
-    pub checksum: u32,
-}
-
-/// Write-ahead log that persists records to a file.
-pub struct WriteAheadLog {
-    path: PathBuf,
-    next_seq: AtomicU64,
-}
-
-impl WriteAheadLog {
-    /// Create a new WAL at the given file path.
-    pub fn new(path: PathBuf) -> Self {
-        Self {
-            path,
-            next_seq: AtomicU64::new(1),
-        }
-    }
-
-    /// Initialize the WAL (create file if needed, load sequence counter).
-    pub async fn initialize(&self) -> StorageResult<()> {
-        if let Some(parent) = self.path.parent() {
-            tokio::fs::create_dir_all(parent).await?;
-        }
-        if self.path.exists() {
-            // Load existing records to find the max sequence number.
-            let records = self.read_all().await?;
-            if let Some(max_seq) = records.iter().map(|r| r.seq).max() {
-                self.next_seq.store(max_seq + 1, Ordering::SeqCst);
-            }
-        }
-        debug!("WAL initialized at {:?}", self.path);
-        Ok(())
-    }
-
-    /// Append a record to the WAL. Returns the sequence number.
-    pub async fn append(
-        &self,
-        op: WalOp,
-        db: &str,
-        collection: &str,
-        document_id: &str,
-        document: Option<Document>,
-    ) -> StorageResult<u64> {
-        let seq = self.next_seq.fetch_add(1, Ordering::SeqCst);
-
-        // Compute checksum over the payload.
-        let payload = serde_json::json!({
-            "op": op,
-            "db": db,
-            "collection": collection,
-            "document_id": document_id,
-        });
-        let payload_bytes = serde_json::to_vec(&payload)?;
-        let checksum = crc32fast::hash(&payload_bytes);
-
-        let record = WalRecord {
-            seq,
-            op,
-            db: db.to_string(),
-            collection: collection.to_string(),
-            document_id: document_id.to_string(),
-            document,
-            committed: false,
-            checksum,
-        };
-
-        let line = serde_json::to_string(&record)?;
-        let mut file = tokio::fs::OpenOptions::new()
-            .create(true)
-            .append(true)
-            .open(&self.path)
-            .await?;
-        file.write_all(line.as_bytes()).await?;
-        file.write_all(b"\n").await?;
-        file.flush().await?;
-
-        Ok(seq)
-    }
-
-    /// Mark a WAL record as committed by rewriting the file.
-    pub async fn mark_committed(&self, seq: u64) -> StorageResult<()> {
-        let mut records = self.read_all().await?;
-        for record in &mut records {
-            if record.seq == seq {
-                record.committed = true;
-            }
-        }
-        self.write_all(&records).await
-    }
-
-    /// Read all WAL records.
-    pub async fn read_all(&self) -> StorageResult<Vec<WalRecord>> {
-        if !self.path.exists() {
-            return Ok(vec![]);
-        }
-        let data = tokio::fs::read_to_string(&self.path).await?;
-        let mut records = Vec::new();
-        for line in data.lines() {
-            if line.trim().is_empty() {
-                continue;
-            }
-            match serde_json::from_str::<WalRecord>(line) {
-                Ok(record) => records.push(record),
-                Err(e) => {
-                    warn!("skipping corrupt WAL record: {e}");
-                }
-            }
-        }
-        Ok(records)
-    }
-
-    /// Get all uncommitted records (for replay during recovery).
-    pub async fn uncommitted(&self) -> StorageResult<Vec<WalRecord>> {
-        let records = self.read_all().await?;
-        Ok(records.into_iter().filter(|r| !r.committed).collect())
-    }
-
-    /// Truncate the WAL, removing all committed records.
-    pub async fn truncate_committed(&self) -> StorageResult<()> {
-        let records = self.read_all().await?;
-        let uncommitted: Vec<_> = records.into_iter().filter(|r| !r.committed).collect();
-        self.write_all(&uncommitted).await
-    }
-
-    /// Clear the entire WAL.
-    pub async fn clear(&self) -> StorageResult<()> {
-        if self.path.exists() {
-            tokio::fs::write(&self.path, "").await?;
-        }
-        self.next_seq.store(1, Ordering::SeqCst);
-        Ok(())
-    }
-
-    /// Write all records to the WAL file (overwrites).
-    async fn write_all(&self, records: &[WalRecord]) -> StorageResult<()> {
-        let mut content = String::new();
-        for record in records {
-            let line = serde_json::to_string(record)?;
-            content.push_str(&line);
-            content.push('\n');
-        }
-        tokio::fs::write(&self.path, content).await?;
-        Ok(())
-    }
-}
@@ -13,7 +13,7 @@ use tokio_util::sync::CancellationToken;
 use rustdb_config::{RustDbOptions, StorageType};
 use rustdb_wire::{WireCodec, OP_QUERY};
 use rustdb_wire::{encode_op_msg_response, encode_op_reply_response};
-use rustdb_storage::{StorageAdapter, MemoryStorageAdapter, FileStorageAdapter};
+use rustdb_storage::{StorageAdapter, MemoryStorageAdapter, FileStorageAdapter, OpLog};
 // IndexEngine is used indirectly via CommandContext
 use rustdb_txn::{TransactionEngine, SessionEngine};
 use rustdb_commands::{CommandRouter, CommandContext};
@@ -56,6 +56,7 @@ impl RustDb {
            sessions: Arc::new(SessionEngine::new(30 * 60 * 1000, 60 * 1000)),
            cursors: Arc::new(DashMap::new()),
            start_time: std::time::Instant::now(),
+            oplog: Arc::new(OpLog::new()),
        });

        let router = Arc::new(CommandRouter::new(ctx.clone()));
@@ -166,6 +167,11 @@ impl RustDb {
    pub fn connection_uri(&self) -> String {
        self.options.connection_uri()
    }
+
+    /// Get a reference to the shared command context (for management IPC access to oplog, storage, etc.).
+    pub fn ctx(&self) -> &Arc<CommandContext> {
+        &self.ctx
+    }
 }

 /// Handle a single client connection using the wire protocol codec.
@@ -139,7 +139,12 @@ async fn handle_request(
        "start" => handle_start(&id, &request.params, db).await,
        "stop" => handle_stop(&id, db).await,
        "getStatus" => handle_get_status(&id, db),
-        "getMetrics" => handle_get_metrics(&id, db),
+        "getMetrics" => handle_get_metrics(&id, db).await,
+        "getOpLog" => handle_get_oplog(&id, &request.params, db),
+        "getOpLogStats" => handle_get_oplog_stats(&id, db),
+        "revertToSeq" => handle_revert_to_seq(&id, &request.params, db).await,
+        "getCollections" => handle_get_collections(&id, &request.params, db).await,
+        "getDocuments" => handle_get_documents(&id, &request.params, db).await,
        _ => ManagementResponse::err(id, format!("Unknown method: {}", request.method)),
    }
 }
@@ -223,18 +228,333 @@ fn handle_get_status(
    }
 }

-fn handle_get_metrics(
+async fn handle_get_metrics(
    id: &str,
    db: &Option<RustDb>,
 ) -> ManagementResponse {
    match db.as_ref() {
-        Some(_d) => ManagementResponse::ok(
-            id.to_string(),
-            serde_json::json!({
-                "connections": 0,
-                "databases": 0,
-            }),
-        ),
+        Some(d) => {
+            let ctx = d.ctx();
+            let db_list = ctx.storage.list_databases().await.unwrap_or_default();
+            let mut total_collections = 0u64;
+            for db_name in &db_list {
+                if let Ok(colls) = ctx.storage.list_collections(db_name).await {
+                    total_collections += colls.len() as u64;
+                }
+            }
+            let oplog_stats = ctx.oplog.stats();
+            let uptime_secs = ctx.start_time.elapsed().as_secs();
+
+            ManagementResponse::ok(
+                id.to_string(),
+                serde_json::json!({
+                    "databases": db_list.len(),
+                    "collections": total_collections,
+                    "oplogEntries": oplog_stats.total_entries,
+                    "oplogCurrentSeq": oplog_stats.current_seq,
+                    "uptimeSeconds": uptime_secs,
+                }),
+            )
+        }
        None => ManagementResponse::err(id.to_string(), "Server is not running".to_string()),
    }
 }
+
+fn handle_get_oplog(
+    id: &str,
+    params: &serde_json::Value,
+    db: &Option<RustDb>,
+) -> ManagementResponse {
+    let d = match db.as_ref() {
+        Some(d) => d,
+        None => return ManagementResponse::err(id.to_string(), "Server is not running".to_string()),
+    };
+
+    let ctx = d.ctx();
+    let since_seq = params.get("sinceSeq").and_then(|v| v.as_u64()).unwrap_or(1);
+    let limit = params.get("limit").and_then(|v| v.as_u64()).unwrap_or(100) as usize;
+    let filter_db = params.get("db").and_then(|v| v.as_str());
+    let filter_coll = params.get("collection").and_then(|v| v.as_str());
+
+    let mut entries = ctx.oplog.entries_since(since_seq);
+
+    // Apply filters.
+    if let Some(fdb) = filter_db {
+        entries.retain(|e| e.db == fdb);
+    }
+    if let Some(fcoll) = filter_coll {
+        entries.retain(|e| e.collection == fcoll);
+    }
+
+    let total = entries.len();
+    entries.truncate(limit);
+
+    // Serialize entries to JSON.
+    let entries_json: Vec<serde_json::Value> = entries
+        .iter()
+        .map(|e| {
+            let doc_json = e.document.as_ref().map(|d| bson_doc_to_json(d));
+            let prev_json = e.previous_document.as_ref().map(|d| bson_doc_to_json(d));
+            serde_json::json!({
+                "seq": e.seq,
+                "timestampMs": e.timestamp_ms,
+                "op": match e.op {
+                    rustdb_storage::OpType::Insert => "insert",
+                    rustdb_storage::OpType::Update => "update",
+                    rustdb_storage::OpType::Delete => "delete",
+                },
+                "db": e.db,
+                "collection": e.collection,
+                "documentId": e.document_id,
+                "document": doc_json,
+                "previousDocument": prev_json,
+            })
+        })
+        .collect();
+
+    ManagementResponse::ok(
+        id.to_string(),
+        serde_json::json!({
+            "entries": entries_json,
+            "currentSeq": ctx.oplog.current_seq(),
+            "totalEntries": total,
+        }),
+    )
+}
+
+fn handle_get_oplog_stats(
+    id: &str,
+    db: &Option<RustDb>,
+) -> ManagementResponse {
+    let d = match db.as_ref() {
+        Some(d) => d,
+        None => return ManagementResponse::err(id.to_string(), "Server is not running".to_string()),
+    };
+
+    let stats = d.ctx().oplog.stats();
+    ManagementResponse::ok(
+        id.to_string(),
+        serde_json::json!({
+            "currentSeq": stats.current_seq,
+            "totalEntries": stats.total_entries,
+            "oldestSeq": stats.oldest_seq,
+            "entriesByOp": {
+                "insert": stats.inserts,
+                "update": stats.updates,
+                "delete": stats.deletes,
+            },
+        }),
+    )
+}
+
+async fn handle_revert_to_seq(
+    id: &str,
+    params: &serde_json::Value,
+    db: &Option<RustDb>,
+) -> ManagementResponse {
+    let d = match db.as_ref() {
+        Some(d) => d,
+        None => return ManagementResponse::err(id.to_string(), "Server is not running".to_string()),
+    };
+
+    let target_seq = match params.get("seq").and_then(|v| v.as_u64()) {
+        Some(s) => s,
+        None => return ManagementResponse::err(id.to_string(), "Missing 'seq' parameter".to_string()),
+    };
+    let dry_run = params.get("dryRun").and_then(|v| v.as_bool()).unwrap_or(false);
+
+    let ctx = d.ctx();
+    let current = ctx.oplog.current_seq();
+
+    if target_seq > current {
+        return ManagementResponse::err(
+            id.to_string(),
+            format!("Target seq {} is beyond current seq {}", target_seq, current),
+        );
+    }
+
+    // Collect entries to revert (from target+1 to current), sorted descending for reverse processing.
+    let mut entries_to_revert = ctx.oplog.entries_range(target_seq + 1, current);
+    entries_to_revert.reverse();
+
+    if dry_run {
+        let entries_json: Vec<serde_json::Value> = entries_to_revert
+            .iter()
+            .map(|e| {
+                serde_json::json!({
+                    "seq": e.seq,
+                    "op": match e.op {
+                        rustdb_storage::OpType::Insert => "insert",
+                        rustdb_storage::OpType::Update => "update",
+                        rustdb_storage::OpType::Delete => "delete",
+                    },
+                    "db": e.db,
+                    "collection": e.collection,
+                    "documentId": e.document_id,
+                })
+            })
+            .collect();
+
+        return ManagementResponse::ok(
+            id.to_string(),
+            serde_json::json!({
+                "dryRun": true,
+                "reverted": entries_to_revert.len(),
+                "entries": entries_json,
+            }),
+        );
+    }
+
+    // Execute revert: process each entry in reverse, using storage directly.
+    let mut reverted = 0u64;
+    let mut errors: Vec<String> = Vec::new();
+
+    for entry in &entries_to_revert {
+        let result = match entry.op {
+            rustdb_storage::OpType::Insert => {
+                // Undo insert -> delete the document.
+                ctx.storage.delete_by_id(&entry.db, &entry.collection, &entry.document_id).await
+            }
+            rustdb_storage::OpType::Update => {
+                // Undo update -> restore the previous document.
+                if let Some(ref prev_doc) = entry.previous_document {
+                    ctx.storage
+                        .update_by_id(&entry.db, &entry.collection, &entry.document_id, prev_doc.clone())
+                        .await
+                } else {
+                    errors.push(format!("seq {}: update entry missing previous_document", entry.seq));
+                    continue;
+                }
+            }
+            rustdb_storage::OpType::Delete => {
+                // Undo delete -> re-insert the previous document.
+                if let Some(ref prev_doc) = entry.previous_document {
+                    ctx.storage
+                        .insert_one(&entry.db, &entry.collection, prev_doc.clone())
+                        .await
+                        .map(|_| ())
+                } else {
+                    errors.push(format!("seq {}: delete entry missing previous_document", entry.seq));
+                    continue;
+                }
+            }
+        };
+
+        match result {
+            Ok(()) => reverted += 1,
+            Err(e) => errors.push(format!("seq {}: {}", entry.seq, e)),
+        }
+    }
+
+    // Truncate the oplog to the target sequence.
+    ctx.oplog.truncate_after(target_seq);
+
+    let mut response = serde_json::json!({
+        "dryRun": false,
+        "reverted": reverted,
+        "targetSeq": target_seq,
+    });
+
+    if !errors.is_empty() {
+        response["errors"] = serde_json::json!(errors);
+    }
+
+    ManagementResponse::ok(id.to_string(), response)
+}
+
+async fn handle_get_collections(
+    id: &str,
+    params: &serde_json::Value,
+    db: &Option<RustDb>,
+) -> ManagementResponse {
+    let d = match db.as_ref() {
+        Some(d) => d,
+        None => return ManagementResponse::err(id.to_string(), "Server is not running".to_string()),
+    };
+
+    let ctx = d.ctx();
+    let filter_db = params.get("db").and_then(|v| v.as_str());
+
+    let databases = match ctx.storage.list_databases().await {
+        Ok(dbs) => dbs,
+        Err(e) => return ManagementResponse::err(id.to_string(), format!("Failed to list databases: {}", e)),
+    };
+
+    let mut collections: Vec<serde_json::Value> = Vec::new();
+
+    for db_name in &databases {
+        if let Some(fdb) = filter_db {
+            if db_name != fdb {
+                continue;
+            }
+        }
+
+        if let Ok(colls) = ctx.storage.list_collections(db_name).await {
+            for coll_name in colls {
+                let count = ctx.storage.count(db_name, &coll_name).await.unwrap_or(0);
+                collections.push(serde_json::json!({
+                    "db": db_name,
+                    "name": coll_name,
+                    "count": count,
+                }));
+            }
+        }
+    }
+
+    ManagementResponse::ok(
+        id.to_string(),
+        serde_json::json!({ "collections": collections }),
+    )
+}
+
+async fn handle_get_documents(
+    id: &str,
+    params: &serde_json::Value,
+    db: &Option<RustDb>,
+) -> ManagementResponse {
+    let d = match db.as_ref() {
+        Some(d) => d,
+        None => return ManagementResponse::err(id.to_string(), "Server is not running".to_string()),
+    };
+
+    let db_name = match params.get("db").and_then(|v| v.as_str()) {
+        Some(s) => s,
+        None => return ManagementResponse::err(id.to_string(), "Missing 'db' parameter".to_string()),
+    };
+    let coll_name = match params.get("collection").and_then(|v| v.as_str()) {
+        Some(s) => s,
+        None => return ManagementResponse::err(id.to_string(), "Missing 'collection' parameter".to_string()),
+    };
+    let limit = params.get("limit").and_then(|v| v.as_u64()).unwrap_or(50) as usize;
+    let skip = params.get("skip").and_then(|v| v.as_u64()).unwrap_or(0) as usize;
+
+    let ctx = d.ctx();
+
+    let all_docs = match ctx.storage.find_all(db_name, coll_name).await {
+        Ok(docs) => docs,
+        Err(e) => return ManagementResponse::err(id.to_string(), format!("Failed to find documents: {}", e)),
+    };
+
+    let total = all_docs.len();
+    let docs: Vec<serde_json::Value> = all_docs
+        .into_iter()
+        .skip(skip)
+        .take(limit)
+        .map(|d| bson_doc_to_json(&d))
+        .collect();
+
+    ManagementResponse::ok(
+        id.to_string(),
+        serde_json::json!({
+            "documents": docs,
+            "total": total,
+        }),
+    )
+}
+
+/// Convert a BSON Document to a serde_json::Value.
+fn bson_doc_to_json(doc: &bson::Document) -> serde_json::Value {
+    // Use bson's built-in relaxed extended JSON serialization.
+    let bson_val = bson::Bson::Document(doc.clone());
+    bson_val.into_relaxed_extjson()
+}
@@ -3,6 +3,6 @@
 */
 export const commitinfo = {
  name: '@push.rocks/smartdb',
-  version: '2.0.0',
+  version: '2.2.0',
  description: 'A MongoDB-compatible embedded database server with wire protocol support, backed by a high-performance Rust engine.'
 }
@@ -7,5 +7,19 @@ export * from './ts_smartdb/index.js';
 export { LocalSmartDb } from './ts_local/index.js';
 export type { ILocalSmartDbOptions, ILocalSmartDbConnectionInfo } from './ts_local/index.js';

+// Export migration
+export { StorageMigrator } from './ts_migration/index.js';
+
 // Export commitinfo
 export { commitinfo };
+
+// Re-export oplog / debug types for convenience
+export type {
+  IOpLogEntry,
+  IOpLogResult,
+  IOpLogStats,
+  IRevertResult,
+  ICollectionInfo,
+  IDocumentsResult,
+  ISmartDbMetrics,
+} from './ts_smartdb/index.js';
@@ -2,6 +2,7 @@ import * as crypto from 'crypto';
 import * as path from 'path';
 import * as os from 'os';
 import { SmartdbServer } from '../ts_smartdb/index.js';
+import { StorageMigrator } from '../ts_migration/index.js';

 /**
 * Connection information returned by LocalSmartDb.start()
@@ -73,6 +74,10 @@ export class LocalSmartDb {
      throw new Error('LocalSmartDb is already running');
    }

+    // Run storage migration before starting the Rust engine
+    const migrator = new StorageMigrator(this.options.folderPath);
+    await migrator.run();
+
    // Use provided socket path or generate one
    this.generatedSocketPath = this.options.socketPath ?? this.generateSocketPath();

@@ -0,0 +1,93 @@
+import * as fs from 'fs';
+import * as path from 'path';
+import { migrateV0ToV1 } from './migrators/v0_to_v1.js';
+
+/**
+ * Detected storage format version.
+ * - v0: Legacy JSON format ({db}/{coll}.json files)
+ * - v1: Bitcask binary format ({db}/{coll}/data.rdb directories)
+ */
+type TStorageVersion = 0 | 1;
+
+/**
+ * StorageMigrator — runs before the Rust engine starts.
+ *
+ * Detects the current storage format version and runs the appropriate
+ * migration chain. The Rust engine only knows the current format (v1).
+ *
+ * Migration is safe: original files are never modified or deleted.
+ * On success, a console hint is printed about which old files can be removed.
+ */
+export class StorageMigrator {
+  private storagePath: string;
+
+  constructor(storagePath: string) {
+    this.storagePath = storagePath;
+  }
+
+  /**
+   * Run any needed migrations. Safe to call even if storage is already current.
+   */
+  async run(): Promise<void> {
+    if (!fs.existsSync(this.storagePath)) {
+      return; // No data yet — nothing to migrate
+    }
+
+    const version = this.detectVersion();
+
+    if (version === 1) {
+      return; // Already current
+    }
+
+    if (version === 0) {
+      console.log(`[smartdb] Detected v0 (JSON) storage format at ${this.storagePath}`);
+      console.log(`[smartdb] Running migration v0 → v1 (Bitcask binary format)...`);
+
+      const deletableFiles = await migrateV0ToV1(this.storagePath);
+
+      if (deletableFiles.length > 0) {
+        console.log(`[smartdb] Migration v0 → v1 complete.`);
+        console.log(`[smartdb] The following old files can be safely deleted:`);
+        for (const f of deletableFiles) {
+          console.log(`[smartdb]   ${f}`);
+        }
+      } else {
+        console.log(`[smartdb] Migration v0 → v1 complete. No old files to clean up.`);
+      }
+    }
+  }
+
+  /**
+   * Detect the storage format version by inspecting the directory structure.
+   *
+   * v0: {db}/{coll}.json files exist
+   * v1: {db}/{coll}/data.rdb directories exist
+   */
+  private detectVersion(): TStorageVersion {
+    const entries = fs.readdirSync(this.storagePath, { withFileTypes: true });
+
+    for (const entry of entries) {
+      if (!entry.isDirectory()) continue;
+
+      const dbDir = path.join(this.storagePath, entry.name);
+      const dbEntries = fs.readdirSync(dbDir, { withFileTypes: true });
+
+      for (const dbEntry of dbEntries) {
+        // v1: subdirectory with data.rdb
+        if (dbEntry.isDirectory()) {
+          const dataRdb = path.join(dbDir, dbEntry.name, 'data.rdb');
+          if (fs.existsSync(dataRdb)) {
+            return 1;
+          }
+        }
+        // v0: .json file (not .indexes.json)
+        if (dbEntry.isFile() && dbEntry.name.endsWith('.json') && !dbEntry.name.endsWith('.indexes.json')) {
+          return 0;
+        }
+      }
+    }
+
+    // Empty or unrecognized — treat as v1 (fresh start)
+    return 1;
+  }
+}
@@ -0,0 +1 @@
+export { StorageMigrator } from './classes.storagemigrator.js';
@@ -0,0 +1,253 @@
+import * as fs from 'fs';
+import * as path from 'path';
+import * as crypto from 'crypto';
+import { BSON } from 'bson';
+
+// ---------------------------------------------------------------------------
+// Binary format constants (must match Rust: record.rs)
+// ---------------------------------------------------------------------------
+
+/** File-level magic: "SMARTDB\0" */
+const FILE_MAGIC = Buffer.from('SMARTDB\0', 'ascii');
+/** Current format version */
+const FORMAT_VERSION = 1;
+/** File type tags */
+const FILE_TYPE_DATA = 1;
+const FILE_TYPE_HINT = 3;
+/** File header total size */
+const FILE_HEADER_SIZE = 64;
+/** Per-record magic */
+const RECORD_MAGIC = 0xDB01;
+/** Per-record header size */
+const RECORD_HEADER_SIZE = 22; // 2 + 8 + 4 + 4 + 4
+
+// ---------------------------------------------------------------------------
+// Binary encoding helpers
+// ---------------------------------------------------------------------------
+
+function writeFileHeader(fileType: number): Buffer {
+  const buf = Buffer.alloc(FILE_HEADER_SIZE, 0);
+  FILE_MAGIC.copy(buf, 0);
+  buf.writeUInt16LE(FORMAT_VERSION, 8);
+  buf.writeUInt8(fileType, 10);
+  buf.writeUInt32LE(0, 11); // flags
+  const now = BigInt(Date.now());
+  buf.writeBigUInt64LE(now, 15);
+  // bytes 23..64 are reserved (zeros)
+  return buf;
+}
+
+function encodeDataRecord(timestamp: bigint, key: Buffer, value: Buffer): Buffer {
+  const keyLen = key.length;
+  const valLen = value.length;
+  const totalSize = RECORD_HEADER_SIZE + keyLen + valLen;
+  const buf = Buffer.alloc(totalSize);
+
+  // Write header fields (without CRC)
+  buf.writeUInt16LE(RECORD_MAGIC, 0);
+  buf.writeBigUInt64LE(timestamp, 2);
+  buf.writeUInt32LE(keyLen, 10);
+  buf.writeUInt32LE(valLen, 14);
+  // CRC placeholder at offset 18..22 (will fill below)
+  key.copy(buf, RECORD_HEADER_SIZE);
+  value.copy(buf, RECORD_HEADER_SIZE + keyLen);
+
+  // CRC32 covers everything except the CRC field itself:
+  // bytes [0..18] + bytes [22..]
+  const crc = crc32(Buffer.concat([
+    buf.subarray(0, 18),
+    buf.subarray(22),
+  ]));
+  buf.writeUInt32LE(crc, 18);
+
+  return buf;
+}
+
+function encodeHintEntry(key: string, offset: bigint, recordLen: number, valueLen: number, timestamp: bigint): Buffer {
+  const keyBuf = Buffer.from(key, 'utf-8');
+  const buf = Buffer.alloc(4 + keyBuf.length + 8 + 4 + 4 + 8);
+  let pos = 0;
+  buf.writeUInt32LE(keyBuf.length, pos); pos += 4;
+  keyBuf.copy(buf, pos); pos += keyBuf.length;
+  buf.writeBigUInt64LE(offset, pos); pos += 8;
+  buf.writeUInt32LE(recordLen, pos); pos += 4;
+  buf.writeUInt32LE(valueLen, pos); pos += 4;
+  buf.writeBigUInt64LE(timestamp, pos);
+  return buf;
+}
+
+// ---------------------------------------------------------------------------
+// CRC32 (matching crc32fast in Rust)
+// ---------------------------------------------------------------------------
+
+const CRC32_TABLE = (() => {
+  const table = new Uint32Array(256);
+  for (let i = 0; i < 256; i++) {
+    let crc = i;
+    for (let j = 0; j < 8; j++) {
+      crc = (crc & 1) ? (0xEDB88320 ^ (crc >>> 1)) : (crc >>> 1);
+    }
+    table[i] = crc;
+  }
+  return table;
+})();
+
+function crc32(data: Buffer): number {
+  let crc = 0xFFFFFFFF;
+  for (let i = 0; i < data.length; i++) {
+    crc = CRC32_TABLE[(crc ^ data[i]) & 0xFF] ^ (crc >>> 8);
+  }
+  return (crc ^ 0xFFFFFFFF) >>> 0;
+}
+
+// ---------------------------------------------------------------------------
+// Migration: v0 (JSON) → v1 (Bitcask binary)
+// ---------------------------------------------------------------------------
+
+interface IKeyDirEntry {
+  offset: bigint;
+  recordLen: number;
+  valueLen: number;
+  timestamp: bigint;
+}
+
+/**
+ * Migrate a storage directory from v0 (JSON-per-collection) to v1 (Bitcask binary).
+ *
+ * - Original .json files are NOT modified or deleted.
+ * - New v1 files are written into {db}/{coll}/ subdirectories.
+ * - Returns a list of old files that can be safely deleted.
+ * - On failure, cleans up any partial new files and throws.
+ */
+export async function migrateV0ToV1(storagePath: string): Promise<string[]> {
+  const deletableFiles: string[] = [];
+  const createdDirs: string[] = [];
+
+  try {
+    const dbEntries = fs.readdirSync(storagePath, { withFileTypes: true });
+
+    for (const dbEntry of dbEntries) {
+      if (!dbEntry.isDirectory()) continue;
+
+      const dbDir = path.join(storagePath, dbEntry.name);
+      const collFiles = fs.readdirSync(dbDir, { withFileTypes: true });
+
+      for (const collFile of collFiles) {
+        if (!collFile.isFile()) continue;
+        if (!collFile.name.endsWith('.json')) continue;
+        if (collFile.name.endsWith('.indexes.json')) continue;
+
+        const collName = collFile.name.replace(/\.json$/, '');
+        const jsonPath = path.join(dbDir, collFile.name);
+        const indexJsonPath = path.join(dbDir, `${collName}.indexes.json`);
+
+        // Target directory
+        const collDir = path.join(dbDir, collName);
+        if (fs.existsSync(collDir)) {
+          // Already migrated
+          continue;
+        }
+
+        console.log(`[smartdb]   Migrating ${dbEntry.name}.${collName}...`);
+
+        // Read the JSON collection
+        const jsonData = fs.readFileSync(jsonPath, 'utf-8');
+        const docs: any[] = JSON.parse(jsonData);
+
+        // Create collection directory
+        fs.mkdirSync(collDir, { recursive: true });
+        createdDirs.push(collDir);
+
+        // Write data.rdb
+        const dataPath = path.join(collDir, 'data.rdb');
+        const fd = fs.openSync(dataPath, 'w');
+
+        try {
+          // File header
+          const headerBuf = writeFileHeader(FILE_TYPE_DATA);
+          fs.writeSync(fd, headerBuf);
+
+          let currentOffset = BigInt(FILE_HEADER_SIZE);
+          const keydir: Map<string, IKeyDirEntry> = new Map();
+          const ts = BigInt(Date.now());
+
+          for (const doc of docs) {
+            // Extract _id
+            let idHex: string;
+            if (doc._id && doc._id.$oid) {
+              idHex = doc._id.$oid;
+            } else if (typeof doc._id === 'string') {
+              idHex = doc._id;
+            } else if (doc._id) {
+              idHex = String(doc._id);
+            } else {
+              // Generate a new ObjectId
+              idHex = crypto.randomBytes(12).toString('hex');
+              doc._id = { $oid: idHex };
+            }
+
+            // Serialize to BSON
+            const bsonBytes = BSON.serialize(doc);
+            const keyBuf = Buffer.from(idHex, 'utf-8');
+            const valueBuf = Buffer.from(bsonBytes);
+
+            const record = encodeDataRecord(ts, keyBuf, valueBuf);
+            fs.writeSync(fd, record);
+
+            keydir.set(idHex, {
+              offset: currentOffset,
+              recordLen: record.length,
+              valueLen: valueBuf.length,
+              timestamp: ts,
+            });
+
+            currentOffset += BigInt(record.length);
+          }
+
+          fs.fsyncSync(fd);
+          fs.closeSync(fd);
+
+          // Write keydir.hint
+          const hintPath = path.join(collDir, 'keydir.hint');
+          const hintFd = fs.openSync(hintPath, 'w');
+          fs.writeSync(hintFd, writeFileHeader(FILE_TYPE_HINT));
+          for (const [key, entry] of keydir) {
+            fs.writeSync(hintFd, encodeHintEntry(key, entry.offset, entry.recordLen, entry.valueLen, entry.timestamp));
+          }
+          fs.fsyncSync(hintFd);
+          fs.closeSync(hintFd);
+
+        } catch (writeErr) {
+          // Clean up on write failure
+          try { fs.closeSync(fd); } catch {}
+          throw writeErr;
+        }
+
+        // Copy indexes.json if it exists
+        if (fs.existsSync(indexJsonPath)) {
+          const destIndexPath = path.join(collDir, 'indexes.json');
+          fs.copyFileSync(indexJsonPath, destIndexPath);
+          deletableFiles.push(indexJsonPath);
+        } else {
+          // Write default _id index
+          const destIndexPath = path.join(collDir, 'indexes.json');
+          fs.writeFileSync(destIndexPath, JSON.stringify([{ name: '_id_', key: { _id: 1 } }], null, 2));
+        }
+
+        deletableFiles.push(jsonPath);
+
+        console.log(`[smartdb]   Migrated ${dbEntry.name}.${collName}: ${docs.length} documents`);
+      }
+    }
+  } catch (err) {
+    // Clean up any partially created directories
+    for (const dir of createdDirs) {
+      try {
+        fs.rmSync(dir, { recursive: true, force: true });
+      } catch {}
+    }
+    throw err;
+  }
+
+  return deletableFiles;
+}
@@ -6,3 +6,14 @@ export type { ISmartdbServerOptions } from './server/SmartdbServer.js';

 // Export bridge for advanced usage
 export { RustDbBridge } from './rust-db-bridge.js';
+
+// Export oplog / debug types
+export type {
+  IOpLogEntry,
+  IOpLogResult,
+  IOpLogStats,
+  IRevertResult,
+  ICollectionInfo,
+  IDocumentsResult,
+  ISmartDbMetrics,
+} from './rust-db-bridge.js';
@@ -3,6 +3,82 @@ import * as path from 'path';
 import * as url from 'url';
 import { EventEmitter } from 'events';

+/**
+ * A single oplog entry returned from the Rust engine.
+ */
+export interface IOpLogEntry {
+  seq: number;
+  timestampMs: number;
+  op: 'insert' | 'update' | 'delete';
+  db: string;
+  collection: string;
+  documentId: string;
+  document: Record<string, any> | null;
+  previousDocument: Record<string, any> | null;
+}
+
+/**
+ * Aggregate oplog statistics.
+ */
+export interface IOpLogStats {
+  currentSeq: number;
+  totalEntries: number;
+  oldestSeq: number;
+  entriesByOp: {
+    insert: number;
+    update: number;
+    delete: number;
+  };
+}
+
+/**
+ * Result of a getOpLog query.
+ */
+export interface IOpLogResult {
+  entries: IOpLogEntry[];
+  currentSeq: number;
+  totalEntries: number;
+}
+
+/**
+ * Result of a revertToSeq command.
+ */
+export interface IRevertResult {
+  dryRun: boolean;
+  reverted: number;
+  targetSeq?: number;
+  entries?: IOpLogEntry[];
+  errors?: string[];
+}
+
+/**
+ * A collection info entry.
+ */
+export interface ICollectionInfo {
+  db: string;
+  name: string;
+  count: number;
+}
+
+/**
+ * Result of a getDocuments query.
+ */
+export interface IDocumentsResult {
+  documents: Record<string, any>[];
+  total: number;
+}
+
+/**
+ * Server metrics.
+ */
+export interface ISmartDbMetrics {
+  databases: number;
+  collections: number;
+  oplogEntries: number;
+  oplogCurrentSeq: number;
+  uptimeSeconds: number;
+}
+
 /**
 * Type-safe command definitions for the RustDb IPC protocol.
 */
@@ -10,7 +86,24 @@ type TSmartDbCommands = {
  start: { params: { config: ISmartDbRustConfig }; result: { connectionUri: string } };
  stop: { params: Record<string, never>; result: void };
  getStatus: { params: Record<string, never>; result: { running: boolean } };
-  getMetrics: { params: Record<string, never>; result: any };
+  getMetrics: { params: Record<string, never>; result: ISmartDbMetrics };
+  getOpLog: {
+    params: { sinceSeq?: number; limit?: number; db?: string; collection?: string };
+    result: IOpLogResult;
+  };
+  getOpLogStats: { params: Record<string, never>; result: IOpLogStats };
+  revertToSeq: {
+    params: { seq: number; dryRun?: boolean };
+    result: IRevertResult;
+  };
+  getCollections: {
+    params: { db?: string };
+    result: { collections: ICollectionInfo[] };
+  };
+  getDocuments: {
+    params: { db: string; collection: string; limit?: number; skip?: number };
+    result: IDocumentsResult;
+  };
 };

 /**
@@ -132,7 +225,38 @@ export class RustDbBridge extends EventEmitter {
    return await this.bridge.sendCommand('getStatus', {} as Record<string, never>) as { running: boolean };
  }

-  public async getMetrics(): Promise<any> {
-    return this.bridge.sendCommand('getMetrics', {} as Record<string, never>);
+  public async getMetrics(): Promise<ISmartDbMetrics> {
+    return this.bridge.sendCommand('getMetrics', {} as Record<string, never>) as Promise<ISmartDbMetrics>;
+  }
+
+  public async getOpLog(params: {
+    sinceSeq?: number;
+    limit?: number;
+    db?: string;
+    collection?: string;
+  } = {}): Promise<IOpLogResult> {
+    return this.bridge.sendCommand('getOpLog', params) as Promise<IOpLogResult>;
+  }
+
+  public async getOpLogStats(): Promise<IOpLogStats> {
+    return this.bridge.sendCommand('getOpLogStats', {} as Record<string, never>) as Promise<IOpLogStats>;
+  }
+
+  public async revertToSeq(seq: number, dryRun = false): Promise<IRevertResult> {
+    return this.bridge.sendCommand('revertToSeq', { seq, dryRun }) as Promise<IRevertResult>;
+  }
+
+  public async getCollections(db?: string): Promise<ICollectionInfo[]> {
+    const result = await this.bridge.sendCommand('getCollections', db ? { db } : {}) as { collections: ICollectionInfo[] };
+    return result.collections;
+  }
+
+  public async getDocuments(
+    db: string,
+    collection: string,
+    limit = 50,
+    skip = 0,
+  ): Promise<IDocumentsResult> {
+    return this.bridge.sendCommand('getDocuments', { db, collection, limit, skip }) as Promise<IDocumentsResult>;
  }
 }
@@ -1,4 +1,14 @@
 import { RustDbBridge } from '../rust-db-bridge.js';
+import { StorageMigrator } from '../../ts_migration/index.js';
+import type {
+  IOpLogEntry,
+  IOpLogResult,
+  IOpLogStats,
+  IRevertResult,
+  ICollectionInfo,
+  IDocumentsResult,
+  ISmartDbMetrics,
+} from '../rust-db-bridge.js';

 /**
 * Server configuration options
@@ -66,6 +76,12 @@ export class SmartdbServer {
      throw new Error('Server is already running');
    }

+    // Run storage migration for file-based storage before starting Rust engine
+    if (this.options.storage === 'file' && this.options.storagePath) {
+      const migrator = new StorageMigrator(this.options.storagePath);
+      await migrator.run();
+    }
+
    const spawned = await this.bridge.spawn();
    if (!spawned) {
      throw new Error(
@@ -156,4 +172,59 @@ export class SmartdbServer {
  get host(): string {
    return this.options.host ?? '127.0.0.1';
  }
+
+  // --- OpLog / Debug API ---
+
+  /**
+   * Get oplog entries, optionally filtered.
+   */
+  async getOpLog(params: {
+    sinceSeq?: number;
+    limit?: number;
+    db?: string;
+    collection?: string;
+  } = {}): Promise<IOpLogResult> {
+    return this.bridge.getOpLog(params);
+  }
+
+  /**
+   * Get aggregate oplog statistics.
+   */
+  async getOpLogStats(): Promise<IOpLogStats> {
+    return this.bridge.getOpLogStats();
+  }
+
+  /**
+   * Revert database state to a specific oplog sequence number.
+   * Use dryRun=true to preview which entries would be reverted.
+   */
+  async revertToSeq(seq: number, dryRun = false): Promise<IRevertResult> {
+    return this.bridge.revertToSeq(seq, dryRun);
+  }
+
+  /**
+   * List all collections across all databases, with document counts.
+   */
+  async getCollections(db?: string): Promise<ICollectionInfo[]> {
+    return this.bridge.getCollections(db);
+  }
+
+  /**
+   * Get documents from a collection with pagination.
+   */
+  async getDocuments(
+    db: string,
+    collection: string,
+    limit = 50,
+    skip = 0,
+  ): Promise<IDocumentsResult> {
+    return this.bridge.getDocuments(db, collection, limit, skip);
+  }
+
+  /**
+   * Get server metrics including database/collection counts and oplog info.
+   */
+  async getMetrics(): Promise<ISmartDbMetrics> {
+    return this.bridge.getMetrics();
+  }
 }
@@ -0,0 +1,120 @@
+import * as plugins from './plugins.js';
+import { files as bundledFiles } from './bundled.js';
+import type { SmartdbServer } from '../ts/index.js';
+
+export interface IDebugServerOptions {
+  /** Port to listen on (default: 4000). */
+  port?: number;
+}
+
+/**
+ * Serves the SmartDB debug UI as a web application with API proxy endpoints.
+ *
+ * @example
+ * ```typescript
+ * import { SmartdbServer } from '@push.rocks/smartdb';
+ * import { SmartdbDebugServer } from '@push.rocks/smartdb/debugserver';
+ *
+ * const db = new SmartdbServer({ storage: 'memory' });
+ * await db.start();
+ *
+ * const debugServer = new SmartdbDebugServer(db, { port: 4000 });
+ * await debugServer.start();
+ * // Open http://localhost:4000
+ * ```
+ */
+export class SmartdbDebugServer {
+  private server: plugins.typedserver.TypedServer;
+  private smartdbServer: SmartdbServer;
+  private port: number;
+
+  constructor(smartdbServer: SmartdbServer, options: IDebugServerOptions = {}) {
+    this.smartdbServer = smartdbServer;
+    this.port = options.port ?? 4000;
+
+    this.server = new plugins.typedserver.TypedServer({
+      cors: true,
+      port: this.port,
+      bundledContent: bundledFiles,
+      spaFallback: true,
+      noCache: true,
+    });
+
+    this.setupApiRoutes();
+  }
+
+  private setupApiRoutes() {
+    const jsonResponse = (data: any, status = 200) =>
+      new Response(JSON.stringify(data), {
+        status,
+        headers: { 'Content-Type': 'application/json' },
+      });
+
+    // Metrics
+    this.server.addRoute('/api/smartdb/metrics', 'GET', async () => {
+      return jsonResponse(await this.smartdbServer.getMetrics());
+    });
+
+    // OpLog stats
+    this.server.addRoute('/api/smartdb/oplog/stats', 'GET', async () => {
+      return jsonResponse(await this.smartdbServer.getOpLogStats());
+    });
+
+    // OpLog entries
+    this.server.addRoute('/api/smartdb/oplog', 'GET', async (ctx) => {
+      const result = await this.smartdbServer.getOpLog({
+        sinceSeq: ctx.query.sinceSeq ? parseInt(ctx.query.sinceSeq) : undefined,
+        limit: ctx.query.limit ? parseInt(ctx.query.limit) : undefined,
+        db: ctx.query.db || undefined,
+        collection: ctx.query.collection || undefined,
+      });
+      return jsonResponse(result);
+    });
+
+    // Collections
+    this.server.addRoute('/api/smartdb/collections', 'GET', async (ctx) => {
+      const collections = await this.smartdbServer.getCollections(ctx.query.db || undefined);
+      return jsonResponse({ collections });
+    });
+
+    // Documents
+    this.server.addRoute('/api/smartdb/documents', 'GET', async (ctx) => {
+      const { db, collection } = ctx.query;
+      if (!db || !collection) {
+        return jsonResponse({ error: 'db and collection required' }, 400);
+      }
+      const limit = parseInt(ctx.query.limit ?? '50');
+      const skip = parseInt(ctx.query.skip ?? '0');
+      const result = await this.smartdbServer.getDocuments(db, collection, limit, skip);
+      return jsonResponse(result);
+    });
+
+    // Revert
+    this.server.addRoute('/api/smartdb/revert', 'GET', async (ctx) => {
+      const { seq, dryRun } = ctx.query;
+      if (!seq) {
+        return jsonResponse({ error: 'seq required' }, 400);
+      }
+      const result = await this.smartdbServer.revertToSeq(
+        parseInt(seq),
+        dryRun === 'true',
+      );
+      return jsonResponse(result);
+    });
+  }
+
+  /**
+   * Start the debug server.
+   */
+  async start(): Promise<void> {
+    await this.server.start();
+    console.log(`SmartDB Debug UI available at http://localhost:${this.port}`);
+  }
+
+  /**
+   * Stop the debug server.
+   */
+  async stop(): Promise<void> {
+    await this.server.stop();
+  }
+}
@@ -0,0 +1,2 @@
+export { SmartdbDebugServer } from './classes.debugserver.js';
+export type { IDebugServerOptions } from './classes.debugserver.js';
@@ -0,0 +1,3 @@
+import * as typedserver from '@api.global/typedserver';
+
+export { typedserver };
@@ -0,0 +1,12 @@
+export { SmartdbDebugUi } from './smartdb-debugui.js';
+
+// Re-export relevant types from the main module
+export type {
+  IOpLogEntry,
+  IOpLogResult,
+  IOpLogStats,
+  IRevertResult,
+  ICollectionInfo,
+  IDocumentsResult,
+  ISmartDbMetrics,
+} from '../ts/index.js';
@@ -0,0 +1,11 @@
+export {
+  DeesElement,
+  customElement,
+  html,
+  css,
+  property,
+  state,
+  cssManager,
+  type TemplateResult,
+  type CSSResult,
+} from '@design.estate/dees-element';
@@ -5,6 +5,8 @@
    "moduleResolution": "NodeNext",
    "esModuleInterop": true,
    "verbatimModuleSyntax": true,
+    "experimentalDecorators": true,
+    "useDefineForClassFields": false,
    "types": ["node"]
  },
  "exclude": [
Author	SHA1	Message	Date
jkunz	4e078b35d4	v2.2.0	2026-04-04 19:49:47 +00:00
jkunz	d8a8259c73	feat(storage): add Bitcask storage migration, binary WAL, and data compaction support	2026-04-04 19:49:47 +00:00
jkunz	9e7ce25b45	v2.1.1	2026-04-02 17:05:07 +00:00
jkunz	b634ee50d1	fix(package): update package metadata	2026-04-02 17:05:07 +00:00
jkunz	5a47c516fd	v2.1.0	2026-04-02 17:02:03 +00:00
jkunz	d34b8673e1	feat(smartdb): add operation log APIs, point-in-time revert support, and a web-based debug dashboard	2026-04-02 17:02:03 +00:00
				`@@ -0,0 +1 @@`
				`export { StorageMigrator } from './classes.storagemigrator.js';`