simply_based/lib/simply_based/store.rb

# frozen_string_literal: true

require 'securerandom'
require 'uri'
require 'couchbase'

module SimplyBased
  # The Couchbase-native data store. Plays the role SimplyCouch's
  # Model::DatabaseInstance played, but speaks the SDK directly: key/value for
  # documents, N1QL/SQL++ for queries, CAS for optimistic concurrency. There is
  # no CouchDB design document or JavaScript map/reduce anywhere.
  #
  # v0 data model: one collection, every document tagged `ruby_class`; queries
  # filter on it and partial GSIs are scoped `WHERE ruby_class = '...'`. A future
  # `collection :name` per-model mode maps each class to its own collection.
  class Store
    attr_reader :cluster, :bucket_name, :scope_name, :collection_name

    def initialize(connection = nil)
      cfg = parse_connection(connection)
      @bucket_name     = cfg[:bucket]
      @scope_name      = cfg[:scope]      || '_default'
      @collection_name = cfg[:collection] || '_default'
      @cluster = ::Couchbase::Cluster.connect(cfg[:conn_str], cfg[:user], cfg[:password])
      @ensured_indexes = {}
      @ensured_indexes_mutex = Mutex.new
    end

    def keyspace
      "`#{bucket_name}`.`#{scope_name}`.`#{collection_name}`"
    end

    # ── single document (KV) ────────────────────────────────────────────────

    def save_document(document, validate = true)
      if document.new?
        create_document(document, validate)
      else
        update_document(document, validate)
      end
    rescue ::Couchbase::Error::CasMismatch, ::Couchbase::Error::DocumentExists
      raise SimplyBased::Conflict.new
    end

    def save_document!(document)
      save_document(document) || raise("Validations failed: #{document.errors.full_messages}")
    end

    def load_document(id)
      raise "Can't load a document without an id (got nil)" if id.nil?
      _track("GET #{id}")
      res = collection.get(id)
      hydrate(res.content, id, res.cas)
    rescue ::Couchbase::Error::DocumentNotFound
      nil
    end

    def destroy_document(document, run_callbacks = true)
      remove = lambda do
        document._deleted = true
        collection.remove(document._id) if document._id
      end
      run_callbacks ? document.run_callbacks(:destroy) { remove.call } : remove.call
      document._id = nil
      document._rev = nil
    rescue ::Couchbase::Error::DocumentNotFound
      document._id = nil; document._rev = nil
    end

    def delete_document(document)
      collection.remove(document._id) if document._id
    rescue ::Couchbase::Error::DocumentNotFound
      nil
    end

    # ── bulk (batched KV) ───────────────────────────────────────────────────

    def bulk_load(ids)
      ids = Array(ids).compact
      return [] if ids.empty?
      _track("BULK_GET #{ids.join(',')}")
      # One batched KV round trip (get_multi); missing ids are dropped, like
      # the CouchDB _bulk_docs behaviour this replaces.
      collection.get_multi(ids).filter_map do |res|
        next unless res.success?
        hydrate(res.content, res.id, res.cas)
      end
    end

    def bulk_save(documents, validate = true)
      documents = Array(documents)
      return { saved: [], invalid: [], failed: [] } if documents.empty?
      invalid = []; saved = []; failed = []; now = Time.now
      documents.each do |document|
        document.database = self
        document.created_at ||= now if document.respond_to?(:created_at) && document.new?
        document.updated_at   = now if document.respond_to?(:updated_at=)
        if validate
          document.errors.clear
          (invalid << document; next) unless valid_document?(document)
        end
        begin
          persist(document)
          document.send(:clear_changes_information) if document.respond_to?(:clear_changes_information, true)
          saved << document
        rescue ::Couchbase::Error::CouchbaseError => e
          failed << [document, e.message]
        end
      end
      { saved: saved, invalid: invalid, failed: failed }
    end

    def bulk_destroy(documents)
      documents = Array(documents)
      return [] if documents.empty?
      with_ids, without_ids = documents.partition(&:_id)
      _track("BULK_REMOVE #{with_ids.map(&:_id).join(',')}") if with_ids.any?
      results = with_ids.any? ? collection.remove_multi(with_ids.map(&:_id)) : []
      rows = with_ids.zip(results).map do |d, res|
        if res.nil? || res.error.nil?
          { 'id' => d._id, 'ok' => true }
        elsif res.error.is_a?(::Couchbase::Error::DocumentNotFound)
          { 'id' => d._id, 'error' => 'not_found' }
        else
          { 'id' => d._id, 'error' => res.error.class.name }
        end
      end
      rows += without_ids.map { { 'id' => nil, 'error' => 'not_found' } }
      documents.each { |d| d._id = nil; d._rev = nil }
      rows
    end

    # ── view specs (finders / count) → N1QL ─────────────────────────────────
    # Handles ModelViewSpec / DeletedModelViewSpec (all_documents, find_by, count).

    def view(spec)
      assert_translatable_view!(spec)
      kls = spec.klass
      vp  = spec.view_parameters
      where = ["c.`ruby_class` = $rc"]
      params = { rc: kls.name }
      where << not_deleted(kls) if deleted_spec?(spec)
      keyexp = key_expr(spec.options[:key])
      where.concat(key_filters(spec.options[:key], keyexp, vp, params))

      results =
        if vp[:reduce]
          [{ 'value' => run_count(where, params) }]
        else
          run_docs(where, params, keyexp, vp).map { |doc| { 'doc' => doc, 'id' => doc._id } }
        end
      processed = spec.process_results({ 'rows' => results })
      assign_db(processed)
      processed
    end

    def first(spec)
      spec.view_parameters = spec.view_parameters.merge(limit: 1)
      Array(view(spec)).first
    end

    # ── associations → N1QL (replaces JS join views) ────────────────────────
    # has_many / has_one / belongs_to: docs of `from_class` whose scalar foreign
    # key column equals `value`.
    def find_by_foreign_key(from_class, column, value, opts = {})
      ensure_index(from_class, [column])
      where = ["c.`ruby_class` = $rc", "c.#{quote_ident(column)} = $v"]
      params = { rc: from_class.name, v: value }
      where << not_deleted(from_class) unless opts[:with_deleted] || !soft_deletes?(from_class)
      run_docs(where, params, 'c.`created_at`', order_opts(opts))
    end

    def count_by_foreign_key(from_class, column, value, opts = {})
      ensure_index(from_class, [column])
      where = ["c.`ruby_class` = $rc", "c.#{quote_ident(column)} = $v"]
      params = { rc: from_class.name, v: value }
      where << not_deleted(from_class) unless opts[:with_deleted] || !soft_deletes?(from_class)
      run_count(where, params)
    end

    # has_and_belongs_to_many: docs of `from_class` whose array column contains value.
    def find_by_array_contains(from_class, column, value, opts = {})
      ensure_array_index(from_class, column)
      where = ["c.`ruby_class` = $rc", "ANY x IN c.#{quote_ident(column)} SATISFIES x = $v END"]
      params = { rc: from_class.name, v: value }
      where << not_deleted(from_class) unless opts[:with_deleted] || !soft_deletes?(from_class)
      run_docs(where, params, 'c.`created_at`', order_opts(opts))
    end

    def count_by_array_contains(from_class, column, value, opts = {})
      ensure_array_index(from_class, column)
      where = ["c.`ruby_class` = $rc", "ANY x IN c.#{quote_ident(column)} SATISFIES x = $v END"]
      params = { rc: from_class.name, v: value }
      where << not_deleted(from_class) unless opts[:with_deleted] || !soft_deletes?(from_class)
      run_count(where, params)
    end

    # ── ancestry (materialized path: path_ids root→self) → N1QL ─────────────
    # No map/reduce build: the tree is expressed directly from the path array.

    # Direct children: docs whose parent (second-to-last path element) is id.
    def tree_children(klass, id, by_property: nil, by_value: nil, order_col: 'position')
      where  = ["c.`ruby_class` = $rc", "ARRAY_LENGTH(c.`path_ids`) >= 2",
                "c.`path_ids`[ARRAY_LENGTH(c.`path_ids`) - 2] = $id"]
      params = { rc: klass.name, id: id }
      _by_property(where, params, by_property, by_value)
      run_tree(where, params, order_col)
    end

    # Whole subtree: docs that carry id somewhere in their path (excluding self).
    def tree_descendants(klass, id, by_property: nil, by_value: nil)
      where  = ["c.`ruby_class` = $rc", "$id IN c.`path_ids`", "META(c).id != $id"]
      params = { rc: klass.name, id: id }
      _by_property(where, params, by_property, by_value)
      run_tree(where, params, nil)
    end

    # Roots: docs whose path is just themselves.
    def tree_roots(klass, by_property: nil, by_value: nil, order_col: 'position')
      where  = ["c.`ruby_class` = $rc", "ARRAY_LENGTH(c.`path_ids`) = 1"]
      params = { rc: klass.name }
      _by_property(where, params, by_property, by_value)
      run_tree(where, params, order_col)
    end

    # ── admin ───────────────────────────────────────────────────────────────

    def create_database!
      query("CREATE PRIMARY INDEX IF NOT EXISTS ON #{keyspace}")
      self
    rescue ::Couchbase::Error::CouchbaseError
      self
    end

    # Clear all documents. Indexes survive (DELETE removes rows, not GSIs), so
    # the @ensured_indexes cache is intentionally kept — between test examples we
    # want to wipe data without rebuilding every association index.
    def drop_database!
      query("DELETE FROM #{keyspace}")
    rescue ::Couchbase::Error::CouchbaseError
      nil
    end

    # ── raw KV primitives (used by Counter / Cache / Lock) ──────────────────

    # Server-side atomic counter (Couchbase binary increment/decrement). Returns
    # the new value. Creates the key at `initial` if absent.
    def counter_incr(key, delta = 1, initial: 0, ttl: nil)
      o = ::Couchbase::Options::Increment.new(delta: delta, initial: initial)
      o.expiry = ttl if ttl
      collection.binary.increment(key, o).content
    end

    def counter_decr(key, delta = 1, initial: 0, ttl: nil)
      o = ::Couchbase::Options::Decrement.new(delta: delta, initial: initial)
      o.expiry = ttl if ttl
      collection.binary.decrement(key, o).content
    end

    def kv_set(key, value, ttl: nil)
      o = ::Couchbase::Options::Upsert.new
      o.expiry = ttl if ttl
      collection.upsert(key, value, o).cas
    end

    # Insert only if absent — returns CAS, or nil if the key already exists.
    def kv_add(key, value, ttl: nil)
      o = ::Couchbase::Options::Insert.new
      o.expiry = ttl if ttl
      collection.insert(key, value, o).cas
    rescue ::Couchbase::Error::DocumentExists
      nil
    end

    def kv_get(key)
      collection.get(key).content
    rescue ::Couchbase::Error::DocumentNotFound
      nil
    end

    def kv_get_cas(key)
      r = collection.get(key)
      [r.content, r.cas]
    rescue ::Couchbase::Error::DocumentNotFound
      nil
    end

    def kv_delete(key, cas: nil)
      o = ::Couchbase::Options::Remove.new
      o.cas = cas if cas
      collection.remove(key, o)
      true
    rescue ::Couchbase::Error::DocumentNotFound, ::Couchbase::Error::CasMismatch
      false
    end

    def query(statement, params = {})
      _track(statement)
      opts = ::Couchbase::Options::Query.new(named_parameters: params)
      opts.scan_consistency = :request_plus # read-your-writes; the suite expects it
      cluster.query(statement, opts)
    end

    private

    # The N1QL translation understands structured view declarations (:key /
    # :keys / start-end ranges / :reduce). CouchDB-era JavaScript escape
    # hatches cannot be honoured — and ignoring them would silently return
    # *unfiltered* results (e.g. a `view :completed, key: :name, conditions:
    # 'doc.completed == true'` would return every document of the class).
    # Fail loudly instead.
    def assert_translatable_view!(spec)
      opts = spec.options || {}
      if opts[:conditions]
        raise SimplyBased::UnsupportedView,
              "view #{spec.klass}##{spec.view_name} declares :conditions (a JavaScript " \
              "fragment); N1QL cannot translate it — express the condition as a key or " \
              "query it via a dedicated finder"
      end
      if opts[:emit_value]
        raise SimplyBased::UnsupportedView,
              "view #{spec.klass}##{spec.view_name} declares :emit_value; the N1QL " \
              "translation always returns full documents"
      end
      if spec.respond_to?(:list_name) && spec.list_name
        raise SimplyBased::UnsupportedView,
              "view #{spec.klass}##{spec.view_name} uses a CouchDB list function " \
              "(#{spec.list_name}); lists have no Couchbase equivalent"
      end
    end

    # Quote an N1QL identifier (field/index name). Identifiers come from the
    # model DSL, not user input, but a backtick smuggled into a property name
    # must never become query text.
    def quote_ident(name)
      n = name.to_s
      raise ArgumentError, "invalid N1QL identifier: #{n.inspect}" if n.include?('`') || n.include?('\\')
      "`#{n}`"
    end

    # Single-quoted N1QL string literal (used only for index WHERE clauses,
    # where named parameters are not allowed).
    def quote_string(value)
      "'#{value.to_s.gsub('\\', '\\\\\\\\').gsub("'", "''")}'"
    end

    # Record a database round-trip for the spec query-count matchers (no-op in
    # production, where $performed_queries is undefined).
    def _track(op)
      $performed_queries << { url: op } if defined?($performed_queries) && $performed_queries
    end

    def collection
      @collection ||= cluster.bucket(bucket_name).scope(scope_name).collection(collection_name)
    end

    # ── N1QL builders ────────────────────────────────────────────────────────

    def run_docs(where, params, order_col, vp)
      sql = +"SELECT c.*, META(c).id AS `__id`, META(c).cas AS `__cas` FROM #{keyspace} AS c WHERE #{where.join(' AND ')}"
      if order_col
        dir = vp[:descending] ? 'DESC' : 'ASC'
        # Within equal keys, order by creation then id: keeps pagination stable
        # and creation-ordered, and (created_at being second-precision) makes
        # ascending and descending exact reverses.
        sql << " ORDER BY #{order_col} #{dir}, c.`created_at` #{dir}, META(c).id #{dir}"
      end
      sql << " LIMIT #{Integer(vp[:limit])}"  if vp[:limit]
      sql << " OFFSET #{Integer(vp[:skip])}"  if vp[:skip]
      query(sql, params).rows.to_a.map { |row| row_to_doc(row) }
    end

    def run_count(where, params)
      sql = "SELECT COUNT(*) AS `value` FROM #{keyspace} AS c WHERE #{where.join(' AND ')}"
      r = query(sql, params).rows.first
      (r && r['value']) || 0
    end

    def key_expr(key)
      case key
      when nil   then nil
      when Array then '[' + key.map { |k| "c.#{quote_ident(k)}" }.join(', ') + ']'
      else "c.#{quote_ident(key)}"
      end
    end

    def key_filters(key_cols, keyexp, vp, params)
      return [] unless keyexp
      if vp.key?(:key)
        # Composite key: compare element-wise so a nil component (e.g. the
        # deleted_at the soft-delete finders append) matches MISSING/NULL rather
        # than a literal null that never equals a missing field.
        if key_cols.is_a?(Array) && vp[:key].is_a?(Array)
          key_cols.each_with_index.map do |col, i|
            v = vp[:key][i]
            if v.nil?
              "(c.#{quote_ident(col)} IS MISSING OR c.#{quote_ident(col)} IS NULL)"
            else
              params[:"k#{i}"] = v
              "c.#{quote_ident(col)} = $k#{i}"
            end
          end
        else
          params[:key] = vp[:key]; ["#{keyexp} = $key"]
        end
      elsif vp.key?(:keys)
        params[:keys] = vp[:keys]; ["#{keyexp} IN $keys"]
      else
        low, high = vp[:descending] ? %i[endkey startkey] : %i[startkey endkey]
        f = []
        (params[:startkey] = vp[low];  f << "#{keyexp} >= $startkey") if vp.key?(low)
        (params[:endkey]   = vp[high]; f << "#{keyexp} <= $endkey")   if vp.key?(high)
        f
      end
    end

    def order_opts(opts)
      { descending: opts[:descending], limit: opts[:limit], skip: opts[:skip] }
    end

    def _by_property(where, params, by_property, by_value)
      # Only partition by the property when a value is supplied. roots/full_tree
      # with no argument span all partitions (e.g. all locales).
      return if by_property.nil? || by_value.nil?
      where << "c.#{quote_ident(by_property)} = $bp"
      params[:bp] = by_value
    end

    def run_tree(where, params, order_col)
      sql = +"SELECT c.*, META(c).id AS `__id`, META(c).cas AS `__cas` FROM #{keyspace} AS c WHERE #{where.join(' AND ')}"
      # Tie-break equal positions by creation time then id, like run_docs.
      # Records habitually share position 0; ordering by position alone left
      # the tie order to the index scan, which flips between runs — this was
      # the actual cause of the "intermittently flaky ancestry" failures
      # (misdiagnosed as a save/CAS race in the original session report).
      sql << " ORDER BY c.#{quote_ident(order_col)} ASC, c.`created_at` ASC, META(c).id ASC" if order_col
      query(sql, params).rows.to_a.map { |row| row_to_doc(row) }
    end

    def not_deleted(kls)
      a = quote_ident(kls.soft_delete_attribute)
      "(c.#{a} IS MISSING OR c.#{a} IS NULL)"
    end

    def soft_deletes?(kls)
      kls.respond_to?(:soft_deleting_enabled?) && kls.soft_deleting_enabled?
    end

    def deleted_spec?(spec)
      spec.class.name.end_with?('DeletedModelViewSpec')
    end

    # Lazily create the GSIs an association needs — the N1QL analogue of
    # SimplyCouch creating a design document on first view query.
    def ensure_index(kls, columns)
      key = [kls.name, columns].inspect
      cols = columns.map { |col| quote_ident(col) }.join(', ')
      ensure_index_once(key, "sb_#{kls.name.gsub(/\W/, '_').downcase}_#{columns.join('_')}"[0, 60],
                        "(#{cols})", kls)
    end

    def ensure_array_index(kls, column)
      key = [kls.name, :array, column].inspect
      ensure_index_once(key, "sb_#{kls.name.gsub(/\W/, '_').downcase}_arr_#{column}"[0, 60],
                        "(DISTINCT ARRAY x FOR x IN #{quote_ident(column)} END)", kls)
    end

    def ensure_index_once(key, name, index_expr, kls)
      # The mutex only guards the bookkeeping hash (Store instances are shared
      # across threads); CREATE INDEX itself is IF NOT EXISTS, so a lost race
      # costs one redundant statement, never a broken index.
      @ensured_indexes_mutex.synchronize { return if @ensured_indexes[key] }
      query("CREATE INDEX #{quote_ident(name)} IF NOT EXISTS ON #{keyspace}#{index_expr} WHERE `ruby_class` = #{quote_string(kls.name)}")
      @ensured_indexes_mutex.synchronize { @ensured_indexes[key] = true }
    rescue ::Couchbase::Error::CouchbaseError => e
      # Still mark as ensured to avoid retry storms, but say so: a swallowed
      # failure here means every later query on this association may fall back
      # to a primary-index scan.
      warn "SimplyBased: could not ensure index #{name} on #{keyspace} " \
           "(#{e.class}: #{e.message}); queries may fall back to a primary scan"
      @ensured_indexes_mutex.synchronize { @ensured_indexes[key] = true }
    end

    # ── persistence helpers ──────────────────────────────────────────────────

    def create_document(document, validate)
      document.database = self
      if validate
        document.errors.clear
        return false if false == document.run_callbacks(:validation) do
          return false if false == document.run_callbacks(:validation_on_create) do
            return false unless valid_document?(document)
          end
        end
      end
      return false if false == document.run_callbacks(:save) do
        return false if false == document.run_callbacks(:create) do
          document._id ||= SecureRandom.uuid
          res = collection.insert(document._id, document.to_hash)
          document._rev = res.cas.to_s
        end
      end
      true
    end

    def update_document(document, validate)
      if validate
        document.errors.clear
        return false if false == document.run_callbacks(:validation) do
          return false if false == document.run_callbacks(:validation_on_update) do
            return false unless valid_document?(document)
          end
        end
      end
      return false if false == document.run_callbacks(:save) do
        return false if false == document.run_callbacks(:update) do
          persist(document) if document.changed?
        end
      end
      true
    end

    def persist(document)
      if document.new?
        document._id ||= SecureRandom.uuid
        _track("INSERT #{document._id}")
        res = collection.insert(document._id, document.to_hash)
      else
        opts = ::Couchbase::Options::Replace.new
        opts.cas = document._rev.to_i if document._rev
        _track("REPLACE #{document._id}")
        res = collection.replace(document._id, document.to_hash, opts)
      end
      document._rev = res.cas.to_s
      document._id
    end

    def valid_document?(document)
      original = document.errors.to_hash
      document.valid?
      original.each { |k, v| Array(v).each { |m| document.errors.add(k, m) } }
      document.errors.empty?
    end

    # ── hydration ────────────────────────────────────────────────────────────

    def row_to_doc(row)
      id  = row['__id'] || row['_id']
      cas = row['__cas'] # the document's CAS is its _rev surrogate
      doc = row.reject { |k, _| k == '__id' || k == '__cas' }
      doc['_id'] ||= id
      hydrate(doc, id, cas || doc['_rev'])
    end

    def hydrate(content, id, rev)
      return content unless content.is_a?(Hash)
      hash = content.dup
      hash['_id'] ||= id
      hash['_rev'] = rev.to_s unless rev.nil?
      model = SimplyBased::DocumentCodec.from_hash(hash)
      model.database = self if model.respond_to?(:database=)
      model
    end

    def assign_db(results)
      return unless results.respond_to?(:each)
      results.each { |d| d.database = self if d.respond_to?(:database=) }
    end

    def parse_connection(connection)
      return connection if connection.is_a?(Hash)
      uri = URI.parse(connection.to_s)
      q = URI.decode_www_form(uri.query.to_s).to_h
      host = [uri.host, (":#{uri.port}" if uri.port)].compact.join
      {
        conn_str:   "#{uri.scheme || 'couchbase'}://#{host}",
        user:       uri.user && URI.decode_www_form_component(uri.user),
        password:   uri.password && URI.decode_www_form_component(uri.password),
        bucket:     uri.path.sub(%r{\A/}, ''),
        scope:      q['scope'],
        collection: q['collection']
      }
    end
  end
end