/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2015-2022 The Fluent Bit Authors
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

#include <fluent-bit/flb_mem.h>
#include <fluent-bit/flb_str.h>
#include <fluent-bit/flb_log.h>
#include <fluent-bit/flb_sqldb.h>

/*
 * Open or create a new database. Note that this function will always try to
 * use an open database and share it handler in as a new context.
 */
struct flb_sqldb *flb_sqldb_open(const char *path, const char *desc,
                                 struct flb_config *config)
{
    int ret;
    struct mk_list *head;
    struct flb_sqldb *db_temp = NULL;
    struct flb_sqldb *db;
    sqlite3 *sdb = NULL;

    db = flb_malloc(sizeof(struct flb_sqldb));
    if (!db) {
        flb_errno();
        return NULL;
    }
    db->parent = NULL;
    db->shared = FLB_FALSE;
    db->users  = 0;

    /*
     * The database handler can be shared across different instances of
     * Fluent Bit. Before to open a new one, try to find a database that
     * is already open.
     */
    mk_list_foreach(head, &config->sqldb_list) {
        db_temp = mk_list_entry(head, struct flb_sqldb, _head);

        /* Only lookup for original database, not contexts already shared */
        if (db_temp->shared == FLB_TRUE) {
            continue;
        }

        if (strcmp(db_temp->path, path) == 0) {
            break;
        }
        db_temp = NULL;
    }

    /* Found a database that can be shared */
    if (db_temp) {
        /* Increase users counter */
        db_temp->users++;

        /* Setup the new context */
        db->handler = db_temp->handler;
        db->shared  = FLB_TRUE;
        db->parent  = db_temp;
    }
    else {
        ret = sqlite3_open(path, &sdb);
        if (ret) {
            flb_error("[sqldb] cannot open database %s", path);
            flb_free(db);
            return NULL;
        }
        db->handler = sdb;
    }

    db->path = flb_strdup(path);
    db->desc = flb_strdup(desc);
    mk_list_add(&db->_head, &config->sqldb_list);

    return db;
}

int flb_sqldb_close(struct flb_sqldb *db)
{
    struct flb_sqldb *parent;

    if (db->shared == FLB_TRUE) {
        parent = db->parent;
        parent->users--;
    }
    else {
        sqlite3_close(db->handler);
    }
    mk_list_del(&db->_head);
    flb_free(db->path);
    flb_free(db->desc);
    flb_free(db);

    return 0;
}

int flb_sqldb_query(struct flb_sqldb *db, const char *sql,
                    int (*callback) (void *, int, char **, char **),
                    void *data)
{
    int ret;
    char *err_msg = NULL;

    ret = sqlite3_exec(db->handler, sql, callback, data, &err_msg);
    if (ret != SQLITE_OK) {
        flb_error("[sqldb] error=%s", err_msg);
        sqlite3_free(err_msg);
        return FLB_ERROR;
    }

    return FLB_OK;
}

int64_t flb_sqldb_last_id(struct flb_sqldb *db)
{
    return sqlite3_last_insert_rowid(db->handler);
}