
package dbmodel

  /** Version 10.12
   *
   *  An ordered collection [[OColl]] is just a collection endowed
   *  with an explicit ordering; i.e. the collection is sorted
   *  according to an associated sorting [[Key]]. It is used mainly
   *  for producing a Synchrony iterator on the underlying collection.
   *
   *  Wong Limsoon
   *  13 May 2023
   */


  object OrderedCollection:

    import scala.language.implicitConversions
    import scala.annotation.targetName
    import scala.util.Using
    import scala.collection.BufferedIterator
    import scala.collection.mutable.Builder
    import dbmodel.RESOURCE.Resource
    import dbmodel.Synchronizable.CBI


    type Bool       = Boolean       
    type Ord[K]     = Ordering[K]
    type BUILDER[B] = Builder[B,Vector[B]]


    /**  Put a copy of [[Key]] here since [[OColl]] is always used with [[Key]]
     */

    type Key[B,K] = dbmodel.KEY.Key[B,K]
    val Key = dbmodel.KEY.Key



    trait OColl[B,K] extends Resource:

      def elems: IterableOnce[B]    /** The collection, set to null if on disk
                                     */

      def key: Key[B,K]             /** [[elems]] is ordered by this [[key]]
                                     */

      def fileElems: CBI[B] = null  /** A hook for [[OFile]]
                                     */

      /** [[OCOL]] is the type and constructor of the inheriting class
       */

      type OCOL[N] <: OColl[B,N]

      protected def OCOL[N](bs: IterableOnce[B], kB: Key[B,N]): OCOL[N]

   
      /** Initialization codes for [[OColl]] constructors to call.
       */

      protected def init() = 
        if elems == null then { }
        else elems match {
          case r: Resource => use(r)
          case _ => 
        }
      

      /** All elements in this collection
       */

      def elemSeq: Vector[B] = elems match 
        case null => fileElems.done { _.toVector }
        case ei: Iterable[B] => ei.toVector
        case ei: Iterator[B] =>
          try ei.asInstanceOf[CBI[B]].done { _.toVector }
          catch _ => ei.toVector
        case _ => Vector()


      /** @return the n-th element in the collection.
       *  It is mainly for testing/checking purpose. I.e., do not 
       *  use this method to access elements; it can be very slow.
       */

      def apply(n: Int): B = elems match
        case null => 
          try fileElems.done { _.drop(n).next() }
          catch { 
            case _: Throwable => 
              throw new NoSuchElementException("OColl.apply")
          }
        case ei: IndexedSeq[B] => ei(n)
        case ei: Iterable[B] => ei.drop(n).head
        case ei: Iterator[B] =>
          try ei.asInstanceOf[CBI[B]].done { _.drop(n).next() }
          catch _ => ei.drop(n).next() 


      
      /** [[cbi]] is an iterator on the underlying collection. 
       *  It "forgets" the ordering information. But it still captures
       *  the dependency information that [[elems]] is its source, to
       *  facilitate auto-destruction of [[elems]] (if it is marked as
       *  "unprotected") when the iterator is closed.
       */

      def cbi: CBI[B] = if elems == null then fileElems else CBI(elems)
  

      /** [[materialized]] converts this transient [[OCBI]] to 
       *  a fully computed [[OColl]]. As such, when the computation
       *  is done, resources held by this transient [[OCBI]] 
       *  can be released.
       */

      def materialized: OCOL[K] = try OCOL(elemSeq, key) finally close()


      /** Methods for reordering the collection by another sorting key
       */

      def assumeOrderedBy[N](ky: Key[B,N]): OCOL[N] = OCOL(elems, ky)


      def ordered: OCOL[K] = orderedBy[K](key)


      def orderedBy[N](ky: Key[B,N]): OCOL[N] =
        OCOL(elemSeq.sortBy(ky.get)(ky.ord), ky)


      def sortedBy[N: Ord](f: B => N): OCOL[N] = orderedBy(Key.asc(f))


      def isOrdered: Bool = cbi.done { it =>
        if !it.hasNext then true 
        else
          var cur = it.next()
          var flag = true
          while (flag && it.hasNext) do
            val nx = it.next()
            flag = key.ord.lteq(key.get(cur), key.get(nx))
            cur = nx
          flag
      }


      def reversed: OCOL[K] = OCOL(elemSeq.reverse, key.reversed)


      /** Comprehension syntax, assuming order is preserved.
       *  Transient perspective, so that results are produced as iterator.
       */

      def filter(f: B => Bool): OCOL[K] = OCOL(cbi.filter(f), key)

      def withFilter(f: B => Bool): OCOL[K] = OCOL(cbi.filter(f), key)


      def map(f: B => B): OCOL[K] = OCOL(cbi.map(f), key)

      def flatMap(f: B => IterableOnce[B]): OCOL[K] = 
        OCOL(cbi.flatMap(f), key)


      def map[C](f: B => C): OSeq[C,Unit] =
        OSeq[C,Unit](cbi.map(f), Key.nullKey[C])

      def flatMap[C](f: B => IterableOnce[C]): OSeq[C,Unit] = 
        OSeq(cbi.flatMap(f), Key.nullKey[C])


      def foreach[U](f: B => U): Unit = cbi.foreach(f)


      /** [[done(f)]] runs [[f(this)]] then release resources
       */

      def done[C](f: this.type => C): C =
        Using.resource(this)(f)


      /** [[doit(f)]] runs [[f(this)]] then let the resulting object
       *  inherit resources from this collection. This is useful for
       *  garbage collecting these resources later.
       */

      def doit[R<:Resource](f: this.type => R): R = f(this).use(this)


      /** [[useOnce]] is a constructor to produce an iterator on [[elems]].
       *  At the end of this iterator, this [[OColl]] is discarded.
       *  Use it with caution because [[this.close()]] usually calls
       *  [[this.cbi.close()]], directly or indirectly, and this may lead
       *  to an infinite loop.
       */

      def useOnce: OCOL[K] = OCOL(cbi.userev(this), key)
  


      /** Method to merge this collection with other [[OColl]]. 
       *  Assume these [[OColl]] are sorted with the same key
       *  as this [[OColl]]. 
       */

      def mergedWith(fs: OColl[B,K]*): OCOL[K] =
        import scala.collection.mutable.PriorityQueue
        type BI = CBI[B]
        val cmp = key.ord.reverse.compare _
        val get = key.get
        val ord = new Ordering[BI] {
          override def compare(x: BI, y: BI) = cmp(get(x.head), get(y.head))
        }
        val cfs: Seq[BI] = this.cbi +: fs.map { _.cbi }
        val efs = cfs filter { _.hasNext }
        val hpq = PriorityQueue(efs: _*)(ord)
        val it = new Iterator[B] {
          def hasNext = !hpq.isEmpty
          def next() = 
            val ef = hpq.dequeue()
            val hd = ef.next()
            if ef.hasNext then hpq.addOne(ef)
            hd
        }
        // val cit  = CBI(it).use(this).use(cfs: _*).use(fs: _*)
        val cit  = CBI(it)
        OCOL(cit, key).use(this).use(cfs: _*).use(fs: _*)



      /** Methods for grouping items in this [[OColl]] by their keys.
       *  Assume this [[OColl]] is sorted by its [[key]] already.
       *  Apply an aggregate function to each group.
       *
       *  @param e     is initial/zero value for the aggregate function.
       *  @param iter  defines the aggregate function; it is called whenever
       *               an item is added to the group being aggregated on.
       *  @param done  is function transforming the aggregation result.
       */

      def clustered: OSeq[(K,Vector[B]),K] = 
        val e = Vector.newBuilder[B]
        val iter = (b: B, acc: BUILDER[B]) => acc += b  
        val done = (acc: BUILDER[B]) => 
          val tmp = acc.result
          acc.clear 
          tmp
        clusteredFold(e, iter, done)


      def clusteredFold[C](e: C, iter: (B,C) => C): OSeq[(K,C),K] =
        val done = (x: C) => x
        clusteredFold(e, iter, done)


      def clusteredFold[C,D](
              e: C,  
              iter: (B,C) => C, 
              done: C => D)
          : OSeq[(K,D),K] =
        type KD = (K,D)
        val ky  = Key[KD,K](_._1, key.ord)
        val cit = clusteredFold[C,D,KD](e, iter, done, (k,d) => k->d)
        OSeq(cit, ky)


      def clusteredFold[C,D,E](
              e: C,  
              iter: (B,C) => C, 
              done: C => D,
              output: (K,D) => E)
          : CBI[E] =
        cbi.step { it =>
          var acc = e
          val kb  = key.get(it.head)
          while it.hasNext && kb == key.get(it.head) do
            acc = iter(it.next(), acc)
          output(kb, done(acc))
        }



      /** @return the keys of this [[OColl]].
       */

      def keys: OSeq[K,K] = 
        val ky = Key((x: K) => x, key.ord)
        val e = 0
        val iter = (b: B, acc: Int) => acc  
        val done = (acc: Int) => acc
        val output = (k: K, _: Int) => k
        val cit = clusteredFold(e, iter, done, output)
        OSeq(cit, ky)

    end OColl



    /** [[OSeq]] represents all in-memory ordered collections.
     */

    trait OSeq[B,K] extends OColl[B,K]:
      type OCOL[N] = OSeq[B,N]
      def OCOL[N](es: IterableOnce[B], ky: Key[B,N]): OCOL[N] = OSeq(es, ky)
      def elems: IterableOnce[B] 
      def key: Key[B,K]
    

    object OSeq: 

      def apply[B,K](bs: IterableOnce[B], kB: Key[B,K]): OSeq[B,K] =
        new OSeq[B,K] { val (elems, key) = (bs, kB); init() }


      /** Some additional constructors, so that programmers do not
       *  need to provide ordering explicitly.
       */

      def apply[B,K: Ord](bs: IterableOnce[B], f: B => K): OSeq[B,K] =
        OSeq[B,K](bs, Key.asc(f))


      def apply[B: Ord](bs: B*): OSeq[B,B] =
        val kB: Key[B,B] = Key.asc[B,B](x => x)
        val sorted = bs.sortBy(x => x)
        OSeq[B,B](sorted, kB)

    end OSeq 
      


    /** Implicit conversions between [[OSeq]], [[CBI]] and [[IterableOnce]].
     */

    given ocoll2cbi[B,K]: Conversion[OColl[B,K], CBI[B]] = _.cbi

    given it2oseq[B]: Conversion[IterableOnce[B], OSeq[B,Unit]] =
      OSeq[B,Unit](_, (x:B) => ())


    extension [B](bs: IterableOnce[B])
      def oseq: OSeq[B,Unit] = OSeq(bs, (x:B) => ())
      def oseq[K](kB: Key[B,K]): OSeq[B,K] = OSeq(bs, kB)
      def oseq[K:Ord](f: B => K): OSeq[B,K] = OSeq(bs, f)

  end OrderedCollection
    


