/*
 * redis.c  (C) SOFNEC
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>

#include "hiredis.h"
#include "putil.h"

#include <azst.h>

//#define DEBUG

#define AZREDIS_VERSION   "0.8"

#define MAX_COMMAND_ARGS_NUM  200


#ifdef UNIFY_RETURN_ATOM
#define UNIFY_ATOM_OR_LIST(Env, t, str) \
  unify_atom((t), PutAtom((Env), (str)))
#else
#define UNIFY_ATOM_OR_LIST(Env, t, str) \
  (! az_cstring_to_list((Env), (str), strlen(str), (t)))
#endif

extern pred P1_redis_is_disconnected(Frame *Env);
extern pred P4_redis_connect(Frame *Env);
extern pred P1_redis_free_context(Frame *Env);
extern pred P4_redis_command(Frame *Env);
extern pred P4_redis_command_hate_integer(Frame *Env);
extern pred P1_redis_version(Frame *Env);

static BASEINT REDIS_REPLY_STRING_ATOM;
static BASEINT REDIS_REPLY_ARRAY_ATOM;
static BASEINT REDIS_REPLY_INTEGER_ATOM;
static BASEINT REDIS_REPLY_NIL_ATOM;
static BASEINT REDIS_REPLY_STATUS_ATOM;
static BASEINT REDIS_REPLY_ERROR_ATOM;
static BASEINT REDIS_REPLY_UNKNOWN_ATOM;
static BASEINT REDIS_VERSION_ATOM;

static st_table* ContextTable;


#ifdef WIN32
  __declspec(dllexport) extern int initiate_redis(void)
#else
  extern int initiate_redis(void)
#endif
{
  char buf[256];

  if (ContextTable != 0)
    st_free_table(ContextTable);

  ContextTable = st_init_numtable();

  put_bltn("redis_connect\0",              4, P4_redis_connect);
  put_bltn("redis_free_context\0",         1, P1_redis_free_context);
  put_bltn("redis_command\0",              4, P4_redis_command);
  put_bltn("redis_command_hate_integer\0", 4, P4_redis_command_hate_integer);
  put_bltn("redis_version\0",              1, P1_redis_version);

  /* !!! redis_is_disconnected() must be called after redis_command() fail. */
  put_bltn("redis_is_disconnected\0",      1, P1_redis_is_disconnected);

  REDIS_REPLY_STRING_ATOM  = PutAtom((Frame* )NULL, "redis_REPLY_STRING");
  REDIS_REPLY_ARRAY_ATOM   = PutAtom((Frame* )NULL, "redis_REPLY_ARRAY");
  REDIS_REPLY_INTEGER_ATOM = PutAtom((Frame* )NULL, "redis_REPLY_INTEGER");
  REDIS_REPLY_NIL_ATOM     = PutAtom((Frame* )NULL, "redis_REPLY_NIL");
  REDIS_REPLY_STATUS_ATOM  = PutAtom((Frame* )NULL, "redis_REPLY_STATUS");
  REDIS_REPLY_ERROR_ATOM   = PutAtom((Frame* )NULL, "redis_REPLY_ERROR");
  REDIS_REPLY_UNKNOWN_ATOM = PutAtom((Frame* )NULL, "redis_REPLY_UNKNOWN");


  sprintf(buf, "redis-ext: %s, hiredis: %d.%d.%d", AZREDIS_VERSION,
          HIREDIS_MAJOR, HIREDIS_MINOR, HIREDIS_PATCH);
  REDIS_VERSION_ATOM = PutAtom((Frame* )NULL, buf);

  return 1;
}


static int
make_element(redisReply* e, Frame* Env, TERM* t, int hate_integer)
{
  if (e->type == REDIS_REPLY_INTEGER) {
    /* Probably, never come here.*/
    if (hate_integer != 0) {
      char buf[30];
      sprintf(buf, "%lld", e->integer);
      if (UNIFY_ATOM_OR_LIST(Env, t, buf) == 0)
        return -1;
    }
    else {
      if (! UnifyInt(t, (BASEINT )(e->integer)))
        return -1;
    }
  }
  else {
    if (e->len == 0) {
      if (UnifyAtom(t, ATOM_NIL) == 0)
        return -1;
    }
    else {
      if (UNIFY_ATOM_OR_LIST(Env, t, e->str) == 0)
        return -1;
    }
  }

  return 0;
}

static int
make_list_from_reply_array(Frame* Env, TERM* t, redisReply* r, int hate_integer)
{
#if 0
  int i;
  TERM *tcar, *tcdr, *head, *tsave;

  if (r->type != REDIS_REPLY_ARRAY)
    return -1;

  tsave = next_var_cell;
  head = t;
  for (i = 0; i < r->elements; i++) {
    redisReply* e = r->element[i];

    MakeUndef(Env); tcar = next_var_cell - 1;
    if (make_element(e, Env, tcar, hate_integer) != 0) {
      goto fail;
    }

    MakeUndef(Env); tcdr = next_var_cell - 1;
    if (! UnifyCons(Env, head, tcar, tcdr))
      goto fail;

    head = tcdr;
  }

  if (UnifyAtom(head, ATOM_NIL) == 0) {
  fail:
    next_var_cell = tsave;
    return -2;
  }

  next_var_cell = tsave;

#else
  int len = r->elements;

  if (len == 0) {
    if (UnifyAtom(t, ATOM_NIL) == 0)
      return -2;
  }
  else {
    int i;
    int rec_size, gvar_size;
    TERM *List_top, *x;

    rec_size = len;
    gvar_size = (BASEINT )(gvar_bottom - next_gvar_cell)/2;	
    if (gvar_size < rec_size) {
      P0_s_greclaim(Env);
      gvar_size = (BASEINT)(gvar_bottom - next_gvar_cell)/2;	
      if (gvar_size < rec_size) {
        return -4;
      }
    }

    List_top = next_gvar_cell;

    for (i = 0; i < r->elements; i++) {
      //redisReply* e = r->element[i];

      SETTAG(next_gvar_cell, list_tag);
      BODY(next_gvar_cell) = next_gvar_cell+1;
      next_gvar_cell++;

      GVCLEAR(next_gvar_cell);
      //next_gvar_cell++;
      //if (make_element(e, Env, next_gvar_cell - 1, hate_integer) != 0) {
      //  return -5;
      //}

      next_gvar_cell++;
    }
    SETTAG(next_gvar_cell, atom_tag);
    INT_BODY(next_gvar_cell++) = ATOM_NIL;

    i = 0;
    x = List_top;
    while (IsCons(x)) {
      redisReply* e = r->element[i];
      REALVALUE(x);
      x = BODY(x);
      if (make_element(e, Env, x, hate_integer) != 0) {
        return -5;
      }

      x++;
      i++;
    }


    if (unify(List_top, t) == 0) {
      return -6;
    }
  }

#endif

  return 0;
}

static BASEINT
type_to_atom(int type)
{
  BASEINT a;

  switch (type) {
  case REDIS_REPLY_STRING:
    a = REDIS_REPLY_STRING_ATOM;
    break;
  case REDIS_REPLY_ARRAY:
    a = REDIS_REPLY_ARRAY_ATOM;
    break;
  case REDIS_REPLY_INTEGER:
    a = REDIS_REPLY_INTEGER_ATOM;
    break;
  case REDIS_REPLY_NIL:
    a = REDIS_REPLY_NIL_ATOM;
    break;
  case REDIS_REPLY_STATUS:
    a = REDIS_REPLY_STATUS_ATOM;
    break;
  case REDIS_REPLY_ERROR:
    a = REDIS_REPLY_ERROR_ATOM;
    break;
  default:
    a = REDIS_REPLY_UNKNOWN_ATOM;
    break;
  }

  return a;
}

/* ?-redis_connect(+ADDRESS, +PORT, +TIMEOUT(msec.), -CONTEXT). */
extern pred
P4_redis_connect(Frame *Env)
{
  int port, t;
  struct timeval timeout;
  redisContext *ctx;
  char buf[AZ_MAX_ATOM_LENGTH + 1];

#ifdef WIN32
  WSADATA data;
  WSAStartup(MAKEWORD(2,0), &data); 
#endif

  ALL_BEGIN(Env);

  Atom2Asciz(GetAtom(PARG(4,0)), buf);
  if (strlen(buf) >= MAX_ADDR_SIZE)
    YIELD(FAIL);

  port = GetInt(PARG(4,1));

  t = GetInt(PARG(4,2));
  timeout.tv_sec  = t / 1000;
  timeout.tv_usec = (t % 1000) * 1000;

  ctx = redisConnectWithTimeout(buf, port, timeout);
  if (ctx->err != 0) {
    fprintf(stderr, "redis_connect: Connection error: '%s'\r\n", ctx->errstr);
    YIELD(FAIL);
  }
  else {
    int r = UnifyInt(PARG(4,3), (BASEINT )ctx);
    if (r != 0) {
      st_insert(ContextTable, (st_data_t )ctx, (st_data_t )0);
    }
    YIELD(r);
  }
}

/* ?-redis_is_disconnected(+CONTEXT). */
extern pred
P1_redis_is_disconnected(Frame *Env)
{
  ALL_BEGIN(Env);

  redisContext* ctx = (redisContext* )GetInt(PARG(1, 0));
  if (ctx == 0)
    YIELD(FAIL);

#if 0
  fprintf(stderr, "err: %d, fd: %d, flags: %d\n", ctx->err, ctx->fd, ctx->flags);
  fflush(stderr);
#endif

  if (! st_lookup(ContextTable, (st_data_t )ctx, (st_data_t* )0)) {
    YIELD(FAIL);
  }

  if (ctx->err == REDIS_ERR_EOF)
    YIELD(DET_SUCC);
  else
    YIELD(FAIL);
}

/* ?-redis_free_context(+CONTEXT). */
extern pred
P1_redis_free_context(Frame *Env)
{
  int r;

  ALL_BEGIN(Env);

  redisContext* ctx = (redisContext* )GetInt(PARG(1,0));
  if (ctx != NULL) {
    if (! st_lookup(ContextTable, (st_data_t )ctx, (st_data_t* )0)) {
      YIELD(FAIL);
    }

    redisFree(ctx);

    r = st_delete(ContextTable, (st_data_t* )&ctx, (st_data_t* )0);
    if (r == 0) {
      YIELD(FAIL);
    }
  }
  else
    YIELD(FAIL);

  YIELD(DET_SUCC);
}


static void clear_command_list(int cn, const char* cs[])
{
  int i;

  for (i = 0; i < cn; i++) {
    if (cs[i] != NULL)
      free((void* )cs[i]);
  }
}

static int prolog_list_to_command(Frame* Env, TERM* t, int max, const char* cs[], size_t cl[])
{
  int r, i, len;
  int n = 0;
  TERM* tb;
  char* p;

  while (IsCons(t)) {
    REALVALUE(t);
    t = BODY(t);
    tb = t;
    REALVALUE(tb);

    if (IsAtom(tb)) {
      len = az_get_atom_term_length(Env, tb);
      if (len > 0) {
        if (n >= max) {
          r = -2;
          goto err;
        }

        p = (char* )malloc((size_t )(len + 1));
        if (p != NULL) {
          if (az_term_to_cstring(Env, tb, p, len + 1) >= 0)
            cs[n++] = p;
        }
      }
    }
    else if (pu_redis_is_prolog_string(tb)) {
      if (n >= max) {
        r = -2;
        goto err;
      }

      len = az_term_to_cstring_length(Env, tb);
      p = (char* )malloc((size_t )(len + 1));
      if (p != NULL) {
        if (az_term_to_cstring(Env, tb, p, len + 1) >= 0)
          cs[n++] = p;
      }
    }
    else {
      r = -1;
    err:
      clear_command_list(n, cs);
      return r;
    }

    t++;
  }

  for (i = 0; i < n; i++) {
    cl[i] = strlen(cs[i]);
  }

  return n;
}

static pred
redis_command(Frame *Env, int hate_integer)
{
  redisContext *ctx;
  redisReply *r = NULL;
  char buf[AZ_MAX_ATOM_LENGTH + 1];
  char *command;
  TERM *t1, *t3;
  BASEINT atype;
  int command_malloced = 0;

  ALL_BEGIN(Env);

  ctx = (redisContext* )GetInt(PARG(4,0));

  if (! st_lookup(ContextTable, (st_data_t )ctx, (st_data_t* )0)) {
    YIELD(FAIL);
  }

  command = NULL;

  t1 = PARG(4,1);
  REALVALUE(t1);
  if (IS_ATOM(t1)) {
    buf[0] = '\0';
    command = buf;

    if (GetAtom(t1) != ATOM_NIL) { /* ignore ATOM_NIL => "[]" */
      Atom2Asciz(GetAtom(t1), buf);
      if (strlen(buf) >= AZ_MAX_ATOM_LENGTH)
        YIELD(FAIL);

#ifdef DEBUG
      fprintf(stderr, "redis_command: Atom: %lld, %s\r\n", GetAtom(t1), buf);
#endif

    exec_command_simple:
#ifdef DEBUG
      fprintf(stderr, "redis_command: COMMAND: '%s'\r\n", command);
      fflush(stderr);
#endif

      if (strlen(command) == 0) {
#ifdef DEBUG
        fprintf(stderr, "redis_command: command is empty.\r\n");
#endif
        if (command_malloced != 0)
          free(command);

        YIELD(FAIL);
      }

      r = redisCommand(ctx, command);

      if (command_malloced != 0)
        free(command);
    }
  }
  else if (IS_LIST(t1)) {
    if (pu_redis_is_prolog_string(t1)) {
      int len = az_term_to_cstring_length(Env, t1);
      command = (char* )malloc((size_t )(len + 1));
      if (command == 0)
        YIELD(FAIL);

      az_term_to_cstring(Env, t1, command, len + 1);

      command_malloced = 1;
      goto exec_command_simple;
    }
    else {
      int cn;
      const char* cs[MAX_COMMAND_ARGS_NUM];
      size_t cl[MAX_COMMAND_ARGS_NUM];

      cn = prolog_list_to_command(Env, t1, MAX_COMMAND_ARGS_NUM, cs, cl);
      if (cn <= 0)
        YIELD(FAIL);
      else {
        r = redisCommandArgv(ctx, cn, cs, cl);
        clear_command_list(cn, cs);
      }
    }
  }
  else {
    AZ_ERROR(9); /* Illegal Argument */
  }

  if (r == (void* )0) { /* ex. client timeout */
    YIELD(FAIL);
  }

  atype = type_to_atom(r->type);

  t3 = PARG(4,3);
  REALVALUE(t3);
  switch (r->type) {
  case REDIS_REPLY_ARRAY:
    if (make_list_from_reply_array(Env, t3, r, hate_integer) < 0) {
      goto fail;
    }
    break;

  case REDIS_REPLY_NIL:
    if (UnifyAtom(t3, ATOM_NIL) == 0)
      goto fail;
    break;

  case REDIS_REPLY_INTEGER:
    if (hate_integer != 0) {
      atype = type_to_atom(REDIS_REPLY_STRING);
      sprintf(buf, "%lld", r->integer);
      if (UNIFY_ATOM_OR_LIST(Env, t3, buf) == 0)
        goto fail;
    }
    else {
      if (UnifyInt(t3, (BASEINT )r->integer) == 0)
        goto fail;
    }
    break;

  case REDIS_REPLY_STATUS:
  case REDIS_REPLY_ERROR:
  case REDIS_REPLY_STRING:
    if (r->len == 0) {
      if (UnifyAtom(t3, ATOM_NIL) == 0)
        goto fail;
    }
    else {
      if (UNIFY_ATOM_OR_LIST(Env, t3, r->str) == 0)
        goto fail;
    }
    break;

  default:
    goto fail;
    break;
  }

  if (unify_atom(PARG(4,2), atype) == 0) {
  fail:
    freeReplyObject(r);
    YIELD(FAIL);
  }

  freeReplyObject(r);
	YIELD(DET_SUCC);
}

/* ?-redis_command(+CONTEXT, +COMMAND, -RETURN_TYPE, -RETURN_VALUE). */
extern pred
P4_redis_command(Frame *Env)
{
  return redis_command(Env, 0);
}

/* ?-redis_command_hate_integer(+CONTEXT, +COMMAND, -RETURN_TYPE, -RETURN_VALUE). */
extern pred
P4_redis_command_hate_integer(Frame *Env)
{
  return redis_command(Env, 1);
}

/* ?-redis_version(-VERSION). */
extern pred
P1_redis_version(Frame *Env)
{
  ALL_BEGIN(Env);

  if (UnifyAtom(PARG(1,0), REDIS_VERSION_ATOM) == 0)
    YIELD(FAIL);

	YIELD(DET_SUCC);
}
