const{EventEmitter:e}=require("events"),t=require("index-encoder"),i=require("sub-encoder"),s=require("mutexify/promise"),r=require("b4a"),{pipeline:n,Transform:a}=require("streamx"),o=r.alloc(1);class c extends e{constructor(e,t){super(),this.index=e,this.watcher=t,this.node=null,this.watcher.on("error",e=>this.emit("error",e)),this.watcher.on("update",()=>{this._updateNode(),this.emit("update")}),this._updateNode()}_updateNode(){const e=this.watcher.node;this.node=e?{key:e.key,value:e.value}:null}close(){return this.watcher.close()}}class h extends e{constructor(e,t){super(),this.collection=e,this.watcher=t,this.node=null,this.watcher.on("error",e=>this.emit("error",e)),this.watcher.on("update",()=>{this._updateNode(),this.emit("update")}),this._updateNode()}_updateNode(){this.node=this.watcher.node?{_id:this.watcher.node.key,value:this.watcher.node.value}:null}close(){return this.watcher.close()}}class l{constructor(e,t){this.bee=e,this.encodings=t}async get(e){const t=await this.bee.get(e,this.encodings);return t?{key:t.key,value:t.value}:null}async delete(e){await this.bee.del(e,this.encodings)}insert(e,t=o){return this.bee.put(e,t,this.encodings)}async peek(e){const t=await this.bee.peek(e,this.encodings);return t?{key:t.key,value:t.value}:null}createReadStream(e={}){const t=this.bee.createReadStream(e,this.encodings);return n(t,new a({transform:(e,t)=>{t(null,{key:e.key,value:e.value})}}))}createDiffStream(e,t={}){const i=this.bee.createDiffStream(e,{...t,...this.encodings});return n(i,new a({transform:(e,t)=>{t(null,{left:e.left?{key:e.left.key,value:e.left.value}:null,right:e.right?{key:e.right.key,value:e.right.value}:null})}}))}async getAndWatch(e,t){const i=await this.bee.getAndWatch(e,{...t,...this.encodings});return new c(this,i)}watch(e,t){return this.bee.watch(e,{...t,...this.encodings})}transaction(e){return new l(e.bee,this.encodings)}}class u{constructor(e,t,{indexes:i,indexer:s,transaction:r,autoincrement:n}={}){this.sub=e,this.primary=t,this.indexes=i||null,this.indexer=s||null,this.insert=this._locked(this._insert.bind(this)),this.update=this._locked(this._update.bind(this)),this.delete=this._locked(this._delete.bind(this)),this._autoincrement=!!n,this._transaction=r}_locked(e){return async(t,i,s)=>{if(!this._transaction)throw new Error("Can only mutate a collection with a transaction");const r=await this._transaction.writeLock(this);try{return await e(t,i,s)}finally{r()}}}async _autoincrementId(){const e=await this.primary.peek({reverse:!0});return e?e.key+1:0}async _update(e,t,{upsert:i=!1}={}){const s=await this.primary.get(e);if(!s&&!i)throw new Error("Cannot update a record that does not exist");return await this.primary.insert(e,t),this.indexer&&await this.indexer(this._transaction,{id:e,prev:s?.value,next:t}),e}async _insert(e){const t=this._autoincrement?await this._autoincrementId():e._id;return this._update(t,e,{upsert:!0})}async _delete(e){const t=await this.primary.get(e);t&&(await this.primary.delete(e),this.indexer&&await this.indexer(this._transaction,{id:e,prev:t.value,next:null}))}async get(e){const t=await this.primary.get(e);return t?{_id:t.key,value:t.value}:null}async peek(e){const t=await this.primary.peek(e);return t?{_id:t.key,value:t.value}:null}createReadStream(e){const t=this.primary.createReadStream(e);return n(t,new a({transform:(e,t)=>t(null,{_id:e.key,value:e.value})}))}createDiffStream(e,t){const i=this.primary.createDiffStream(e,t);return n(i,new a({transform:(e,t)=>t(null,{left:e.left?{_id:e.left.key,value:e.left.value}:null,right:e.right?{_id:e.right.key,value:e.right.value}:null})}))}async getAndWatch(e,t){const i=await this.primary.getAndWatch(e,t);return new h(this,i)}watch(e,t){return this.primary.watch(e,t)}transaction(e){return new u(this.sub,this.primary.transaction(e),{transaction:e,indexer:this.indexer,autoincrement:this._autoincrement,indexes:this.indexes&&this.indexes.map(t=>t.transaction(e))})}createIndexes(e,t){if(this.indexes||this.indexer)throw new Error("Already created indexes");this.indexes=[],this.indexer=t;let i=1;for(const t of e)if(Array.isArray(t)){const e=[];for(const s of t)e.push(w(this.primary.bee,this.sub,i++,s));this.indexes.push(e)}else this.indexes.push(w(this.primary.bee,this.sub,i++,t));return this.indexes}}class d{constructor(e,{collections:t,sub:r,writable:n,release:a,offset:o=0}={}){this.bee=e,this.sub=r||new i,this.collections=t?t.map(e=>e.transaction(this)):null,this._writable=!!n,this._writeLock=this.collections?null:s(),this._collectionLocks=this._writable?new Map:null,this._release=a,this._offset=o}get version(){return this.bee.version}async writeTransaction(e){const t=await this._writeLock();return new this.constructor(this.bee.batch(e),{collections:this.collections,writable:!0,sub:this.sub,release:t})}readTransaction(e){return new this.constructor(this.bee.batch(e),{collections:this.collections,writable:!1,sub:this.sub})}createCollections(e){if(this.collections)throw new Error("Already created collections");this.collections=[];for(let t=0;t<e.length;t++){const i=this.sub.sub(r.from([t+this._offset])),s={...e[t],transaction:this._writable?this:null},n=w(this.bee,i,0,s);this.collections.push(new u(i,n,s))}return this.collections}async writeLock(e){if(!this._writable)throw new Error("Transaction is not writable");let t=this._collectionLocks.get(e);return t||(t=s(),this._collectionLocks.set(e,t)),t()}async commit(){if(!this._writable)throw new Error("Transaction is not writable");await Promise.all([...this._collectionLocks.values()].map(e=>e())),await this.bee.flush(),this._releaseLock()}async close(){await this.bee.close(),this._releaseLock()}_releaseLock(){const e=this._release;this._release=null,e&&e()}static UINT=t.UINT;static STRING=t.STRING;static BUFFER=t.BUFFER}function w(e,i,s,n){let{keyEncoding:a,valueEncoding:o,autoincrement:c}=n;c&&(a=d.UINT);const h=Array.isArray(a)?new t(a):a,u={keyEncoding:i.sub(r.from([s]),h),valueEncoding:o};return new l(e,u)}module.exports=d;