| 19 | } |
| 20 | |
| 21 | func (w *WriteBatch) Put(key, value []byte) { |
| 22 | if w.err != nil { |
| 23 | return |
| 24 | } |
| 25 | // --- Versioning --- |
| 26 | w.db.versionMu.Lock() |
| 27 | w.db.versions[string(key)]++ |
| 28 | version := w.db.versions[string(key)] |
| 29 | versionBytes := []byte(fmt.Sprintf("%d", version)) |
| 30 | metaKey := append([]byte("_meta:"), key...) |
| 31 | w.db.versionMu.Unlock() |
| 32 | |
| 33 | // --- Prepare local and IPFS operations --- |
| 34 | // Encrypt value for IPFS if needed |
| 35 | ipfsValue := value |
| 36 | if w.db.encryptionKey != nil { |
| 37 | encrypted, err := encrypt(value, w.db.encryptionKey) |
| 38 | if err != nil { |
| 39 | w.err = fmt.Errorf("encrypt failed during batch put: %w", err) |
| 40 | return |
| 41 | } |
| 42 | ipfsValue = encrypted |
| 43 | } |
| 44 | |
| 45 | // Add local operations to the leveldb batch |
| 46 | w.wbatch.Put(key, value) |
| 47 | w.wbatch.Put(metaKey, versionBytes) |
| 48 | |
| 49 | // Add IPFS operations to the pre-commit queue |
| 50 | w.ipfsOps = append(w.ipfsOps, func() error { |
| 51 | return w.db.ipfsDB.Put(w.db.ctx, key, ipfsValue) |
| 52 | }) |
| 53 | w.ipfsOps = append(w.ipfsOps, func() error { |
| 54 | return w.db.ipfsDB.Put(w.db.ctx, metaKey, versionBytes) |
| 55 | }) |
| 56 | |