hypercore-kb / opencog-kb / language-learning / chatbot / queue.scm

;
; queue.scm - Message queue for async learning
;
; Simple file-based queue for decoupling input from learning.
; Query process writes, learning daemon reads and processes.
;

(use-modules (ice-9 format))
(use-modules (ice-9 rdelim))
(use-modules (ice-9 ports))

;; ═══════════════════════════════════════════════════════════════════
;; CONFIGURATION
;; ═══════════════════════════════════════════════════════════════════

(define *queue-dir* "data/queue")
(define *queue-file* "data/queue/pending.txt")
(define *processed-file* "data/queue/processed.txt")

;; ═══════════════════════════════════════════════════════════════════
;; QUEUE OPERATIONS
;; ═══════════════════════════════════════════════════════════════════

(define (ensure-queue-dir)
  "Create queue directory if needed"
  (unless (file-exists? *queue-dir*)
    (mkdir *queue-dir*)))

(define (queue-message text)
  "Add a message to the learning queue (non-blocking)"
  (ensure-queue-dir)
  (let ((port (open-file *queue-file* "a")))
    (display text port)
    (newline port)
    (close-port port)
    #t))

(define (queue-length)
  "Count pending messages"
  (if (file-exists? *queue-file*)
      (let ((port (open-input-file *queue-file*)))
        (let loop ((count 0))
          (let ((line (read-line port)))
            (if (eof-object? line)
                (begin (close-port port) count)
                (loop (+ 1 count))))))
      0))

(define (dequeue-batch max-count)
  "Get up to max-count messages from queue (atomic read+clear)"
  (if (not (file-exists? *queue-file*))
      '()
      (let ((messages '())
            (remaining '()))
        ;; Read all messages
        (let ((port (open-input-file *queue-file*)))
          (let loop ((count 0))
            (let ((line (read-line port)))
              (unless (eof-object? line)
                (if (< count max-count)
                    (set! messages (cons line messages))
                    (set! remaining (cons line remaining)))
                (loop (+ 1 count)))))
          (close-port port))

        ;; Write remaining back (or delete if empty)
        (if (null? remaining)
            (when (file-exists? *queue-file*)
              (delete-file *queue-file*))
            (let ((port (open-output-file *queue-file*)))
              (for-each (lambda (msg)
                          (display msg port)
                          (newline port))
                        (reverse remaining))
              (close-port port)))

        ;; Log processed
        (when (not (null? messages))
          (let ((port (open-file *processed-file* "a")))
            (for-each (lambda (msg)
                        (display msg port)
                        (newline port))
                      messages)
            (close-port port)))

        (reverse messages))))

(define (clear-queue)
  "Clear all pending messages"
  (when (file-exists? *queue-file*)
    (delete-file *queue-file*))
  #t)

;; ═══════════════════════════════════════════════════════════════════
;; STATS
;; ═══════════════════════════════════════════════════════════════════

(define (queue-stats)
  "Get queue statistics"
  (let ((pending (queue-length))
        (processed (if (file-exists? *processed-file*)
                       (let ((port (open-input-file *processed-file*)))
                         (let loop ((count 0))
                           (let ((line (read-line port)))
                             (if (eof-object? line)
                                 (begin (close-port port) count)
                                 (loop (+ 1 count))))))
                       0)))
    (format #f "Queue: ~a pending, ~a processed" pending processed)))