;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;; Provide access to HBase via the Stargate REST API
;;; http://wiki.apache.org/hadoop/Hbase/Stargate
;;;
;;;  Copyright (C) 2015, Andy Bennett, Skipjaq Inc.
;;;  All rights reserved.
;;;
;;;  Redistribution and use in source and binary forms, with or without
;;;  modification, are permitted provided that the following conditions are met:
;;;
;;;  Redistributions of source code must retain the above copyright notice, this
;;;  list of conditions and the following disclaimer.
;;;  Redistributions in binary form must reproduce the above copyright notice,
;;;  this list of conditions and the following disclaimer in the documentation
;;;  and/or other materials provided with the distribution.
;;;  Neither the name of the author nor the names of its contributors may be
;;;  used to endorse or promote products derived from this software without
;;;  specific prior written permission.
;;;
;;;  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
;;;  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
;;;  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
;;;  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR CONTRIBUTORS BE
;;;  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
;;;  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
;;;  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
;;;  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
;;;  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
;;;  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
;;;  POSSIBILITY OF SUCH DAMAGE.
;;;
;;; Andy Bennett <andyjpb@skipjaq.com>, 2015/06
;;;
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(module stargate
	(column?
	 column-family
	 column-qualifier
	 cell?
	 cell-row
	 cell-column
	 cell-value
	 scanner?
	 tables
	 schema
	 scanner-table
	 create-scanner
	 scanner-get-next)

(import chicken scheme)

; Units - http://api.call-cc.org/doc/chicken/language
(use srfi-1 data-structures)

; Eggs - http://wiki.call-cc.org/chicken-projects/egg-index-4.html
(use intarweb uri-common matchable base64)
(use (prefix stargate-lolevel ll:))



; ADT helpers

; Returns a predicate that can be used to determine if something is one of the
; ADTs defined in this module: for now we're using a convention over lists
; rather than scheme records.
(define (stargate-adt? name len)
  (lambda (instance)
    (and
      (list? instance)
      (= (+ 1 len) (length instance)) ; car is name, cdr is list of attributes.
      (eqv? name (first instance)))))



; Stargate ADTs

; Table Scanner ADT

(define (make-scanner table scanner-id)
  `(scanner: ,table ,scanner-id))

(define (scanner? scanner)
  ((stargate-adt? scanner: 2) scanner))

(define scanner-table second)

(define scanner-id    third)


; Column & Cell ADTs
; An HBase cell like a cell in a table. i.e. something at the intersection of a
; row and a column.
; In HBase, a Cell is described by it's rowid, column family and column
; qualifier.

(define (make-column family qualifier)
  `(column: ,family ,qualifier))

; Decode x:y into an HBase column family and column qualifer and poke it into a
; column record.
(define (string->column str)
  (let ((x (string-split str ":" #t)))
    (make-column (first x) (string->blob (second x)))))

(define (column? column)
  ((stargate-adt? column: 2) column))

(define column-family    second)

(define column-qualifier third)


(define (make-cell row column timestamp value)
  (assert (column? column))
  `(cell: ,row ,column ,timestamp ,value))

(define (cell? cell)
  ((stargate-adt? cell: 4) cell))

(define cell-row       second)

(define cell-column    third)

(define cell-timestamp fourth)

(define cell-value     fifth)



; API


; Get a list of table names.

(define (tables)
  (receive (lst uri+resp) (ll:tables)
	   lst))


(define (schema table)
  (let* ((result (ll:schema table))
	 (schema (alist-ref 'ColumnSchema result)))
    (alist-update
      'ColumnSchema
      (vector->list schema)
      result)))




; Table Scanner API

(define (create-scanner table)
  (receive (headers str uri+resp) (ll:create-scanner table #f)
	   (assert (string=? "" str))
	   (let ((resp (second uri+resp))
		 (path (uri-path (header-value 'location headers))))
	     (assert (eqv? 'created (response-status resp)))
	     (match path
	       ((/ (? (cut string=? table <>)) "scanner" id)
		(make-scanner table id))
	       (else
		 (abort
		   (sprintf "Unexpected location in table scanner: ~S" path)))))))



(define (scanner-get-next scanner)
  (assert (scanner? scanner))

  (receive (headers body uri+resp) (ll:scanner-get-next
				     (scanner-table scanner)
				     (scanner-id    scanner))
	   (let ((resp (second uri+resp)))
	     (case (response-code resp)
	       ((200) ; next cell
		; decode the headers x-column x-timestamp x-row content-length
		(let ((column (base64-decode (header-value 'x-column    headers)))
		      (row    (base64-decode (header-value 'x-row       headers)))
		      (ts     (base64-decode (header-value 'x-timestamp headers))))
		  (assert (eqv? 'application/octet-stream (header-value 'content-type headers)))
		  (make-cell
		    (string->blob row)
		    (string->column column)
		    ts
		    body)))
	       ((204) ; end of table
		(assert (eq? #f body))
		#f)
	       (else
		 (abort
		   (sprintf "Unexpected response: ~S\n" (response-code resp))))))))





)