(defpackage :fwoar.crdt
   (:use :cl)
   (:export ))
 (in-package :fwoar.crdt)
 
 (defgeneric crdt-update (op crdt))
 (defgeneric crdt-query (crdt))
 (defgeneric crdt-compare (less greater))
 (defgeneric crdt-merge (id first second))

 (defgeneric crdt-serialize (crdt))
 (defgeneric crdt-deserialize (class state)
   (:method ((class symbol) state)
     (crdt-deserialize (closer-mop:class-prototype (find-class class))
                       state)))

 
 (fw.lu:defclass+ g-counter ()

   ((lock :reader unsafe-lock
          :initform (bt:make-lock "g-counter-lock"))
    (id :reader my-id :initarg :id)

    (size :reader size :initarg :size)
    (payload :reader payload)))

 (defmethod crdt-serialize ((crdt g-counter))
   (with-output-to-string (s)
     (format s "~d" (my-id crdt))
     (loop for v across (bt:with-lock-held ((unsafe-lock crdt))
                          (copy-seq (payload crdt)))
           do
              (format s "|~a" v)
           finally
              (format s "~%"))))
 #+comment
 (progn
   (crdt-serialize (positives *x*))
   )
 (defmethod crdt-deserialize ((crdt g-counter) state)
   (let* ((parts (map 'vector 'parse-integer
                      (fwoar.string-utils:split #\|
                                                (serapeum:trim-whitespace state))))
          (result (g-counter (elt parts 0)
                             (1- (length parts)))))
     (prog1 result
       (with-slots (payload) result
         (setf payload (subseq parts 1))))))

 (defmethod print-object ((o g-counter) s)
   (with-slots (id size) o
     (prin1 `(g-counter :id ,id :size ,size)
            s)))
 (defmethod initialize-instance :after ((crdt g-counter) &key size)
   (with-slots (payload) crdt
     (setf payload (make-array size))))
 (defmethod crdt-update ((op (eql :increment)) (crdt g-counter))

   (bt:with-lock-held ((unsafe-lock crdt))
     (incf (aref (payload crdt)
                 (my-id crdt)))))

 (defmethod crdt-query ((crdt g-counter))
   (reduce #'+
           (payload crdt)))
 (defmethod crdt-compare ((x g-counter)
                          (y g-counter))
   (every #'<=
          (payload x)
          (payload y)))
 (defmethod crdt-merge (id
                        (x g-counter)
                        (y g-counter))
   (assert (= (size x)
              (size y))
           (x y))
   (let ((z (g-counter id (size x))))
     (prog1 z

       (bt:with-lock-held ((unsafe-lock x))
         (bt:with-lock-held ((unsafe-lock y))
           (loop for i from 0 below (size x)
                 do
                    (setf (aref (payload z) i)
                          (max (aref (payload x) i)
                               (aref (payload y) i)))))))))

 
 (fw.lu:defclass+ pn-counter ()
   ((id :reader my-id :initarg :id)
    (size :reader size :initarg :size)
    (positives :reader positives)
    (negatives :reader negatives)))
 (defmethod initialize-instance :after ((crdt pn-counter) &key id size)
   (with-slots (positives negatives) crdt
     (setf positives (g-counter id size)
           negatives (g-counter id size))))
 (defmethod print-object ((o pn-counter) s)
   (with-slots (id size positives negatives) o
     (prin1 `(pn-counter :id ,id :size ,size)
            s)))
 (defmethod payload ((object pn-counter))
   (with-slots (positives negatives) object
     (list :positives (payload positives)
           :negatives (payload negatives))))
 (defmethod crdt-update ((op (eql :increment)) (crdt pn-counter))
   (crdt-update :increment (positives crdt)))
 (defmethod crdt-update ((op (eql :decrement)) (crdt pn-counter))
   (crdt-update :increment (negatives crdt)))
 (defmethod crdt-query ((crdt pn-counter))
   (- (crdt-query (positives crdt))
      (crdt-query (negatives crdt))))
 (defmethod crdt-compare ((x pn-counter)
                          (y pn-counter))
   (and (crdt-compare (positives x) (positives y))
        (crdt-compare (negatives x) (negatives y))))
 (defmethod crdt-merge (id
                        (x pn-counter)
                        (y pn-counter))
   (assert (= (size x)
              (size y))
           (x y))
   (let ((z (pn-counter id (size x))))
     (prog1 z
       (with-slots (positives negatives) z
         (with-slots ((p-x positives) (n-x negatives)) x
           (with-slots ((p-y positives) (n-y negatives)) y
             (setf positives (crdt-merge id p-x p-y)
                   negatives (crdt-merge id n-x n-y))))))))
 
 (defun make-cluster (size class)
   (values-list
    (loop for id from 0 below size
          collect (make-instance class :id id :size size))))