--- a/Makefile	Fri Oct 17 16:09:35 2008 +0300
+++ b/Makefile	Fri Oct 17 18:53:05 2008 +0300
@@ -20,7 +20,7 @@
 
 # complex modules
 EVSQL_OBJS = obj/evsql/evsql.o obj/evsql/util.o obj/evpq.o
-DBFS_OBJS = obj/dbfs/dbfs.o obj/dbfs/common.o obj/dbfs/core.o obj/dbfs/op_base.o obj/dbfs/dirop.o obj/dirbuf.o obj/dbfs/fileop.o obj/dbfs/attr.o obj/dbfs/link.o
+DBFS_OBJS = obj/dbfs/dbfs.o obj/dbfs/common.o obj/dbfs/core.o obj/dbfs/op_base.o obj/dbfs/trans.o obj/dbfs/dirop.o obj/dirbuf.o obj/dbfs/fileop.o obj/dbfs/attr.o obj/dbfs/link.o
 
 # first target
 all: ${BIN_PATHS}

--- a/doc/fuse_db.sql	Fri Oct 17 16:09:35 2008 +0300
+++ b/doc/fuse_db.sql	Fri Oct 17 18:53:05 2008 +0300
@@ -25,4 +25,20 @@
 CREATE FUNCTION lo_pwrite (IN fd int4, IN buf bytea, IN "off" int4) RETURNS int4 LANGUAGE SQL STRICT AS 'select lo_lseek($1, $3, 0); select lowrite($1, $2);';
 CREATE FUNCTION lo_otruncate (IN oid, IN len int4) RETURNS oid LANGUAGE SQL STRICT AS 'select lo_truncate(lo_open($1, 393216), $2); select $1;';
 
-ALTER TABLE inodes ADD COLUMN symlink varchar(512);
+ALTER TABLE inodes ADD COLUMN link varchar(512);
+
+CREATE SEQUENCE ino_seq START 64;
+ALTER TABLE inodes ALTER COLUMN ino SET DEFAULT nextval('ino_seq'::regclass);
+
+ALTER TABLE file_tree ADD COLUMN ino int4;
+UPDATE file_tree SET ino = inode;
+ALTER TABLE file_tree DROP COLUMN inode;
+
+CREATE FUNCTION dbfs_size (type char, oid, link varchar) RETURNS int4 LANGUAGE SQL STABLE AS $$
+    SELECT CASE $1 
+        WHEN 'LNK' THEN char_length($3) 
+        WHEN 'REG' THEN lo_size($2) 
+        ELSE 0 
+    END;
+$$;
+

--- a/src/dbfs/attr.c	Fri Oct 17 16:09:35 2008 +0300
+++ b/src/dbfs/attr.c	Fri Oct 17 18:53:05 2008 +0300
@@ -60,8 +60,7 @@
         "SELECT"
         " inodes.ino, " DBFS_STAT_COLS
         " FROM inodes"
-        " WHERE inodes.ino = $1::int4"
-        " GROUP BY inodes.ino, inodes.type, inodes.mode, data";
+        " WHERE inodes.ino = $1::int4";
 
     static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {
         EVSQL_PARAM ( UINT32 ),
@@ -137,7 +136,7 @@
         "UPDATE inodes SET"
         " %s%s%s%s ino = ino"
         " WHERE inodes.ino = $5::int4"
-        " RETURNING inodes.ino, " DBFS_STAT_COLS_NOAGGREGATE,
+        " RETURNING inodes.ino, " DBFS_STAT_COLS,
         
         FIELD(to_set, FUSE_SET_ATTR_MODE,   "mode", "$1::int2"),
         FIELD(to_set, FUSE_SET_ATTR_UID,    "uid",  "$2::int4"),

--- a/src/dbfs/core.c	Fri Oct 17 16:09:35 2008 +0300
+++ b/src/dbfs/core.c	Fri Oct 17 18:53:05 2008 +0300
@@ -57,9 +57,8 @@
     const char *sql = 
         "SELECT"
         " inodes.ino, " DBFS_STAT_COLS
-        " FROM file_tree INNER JOIN inodes ON (file_tree.inode = inodes.ino)"
-        " WHERE file_tree.parent = $1::int4 AND file_tree.name = $2::varchar"
-        " GROUP BY inodes.ino, inodes.type, inodes.mode, data";
+        " FROM file_tree INNER JOIN inodes ON (file_tree.ino = inodes.ino)"
+        " WHERE file_tree.parent = $1::int4 AND file_tree.name = $2::varchar";
     
     static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {
         EVSQL_PARAM ( UINT32 ),

--- a/src/dbfs/dbfs.c	Fri Oct 17 16:09:35 2008 +0300
+++ b/src/dbfs/dbfs.c	Fri Oct 17 18:53:05 2008 +0300
@@ -16,6 +16,8 @@
     .setattr        = dbfs_setattr,
     .readlink       = dbfs_readlink,
 
+    .symlink        = dbfs_symlink,
+
     .open           = dbfs_open,
     .read           = dbfs_read,
     .write          = dbfs_write,

--- a/src/dbfs/dbfs.h	Fri Oct 17 16:09:35 2008 +0300
+++ b/src/dbfs/dbfs.h	Fri Oct 17 18:53:05 2008 +0300
@@ -14,8 +14,6 @@
  * Structs and functions shared between all dbfs components
  */
 
-#define SERROR(val) do { (val); goto error; } while(0)
-
 struct dbfs {
     struct event_base *ev_base;
     
@@ -29,8 +27,7 @@
 #define CACHE_TIMEOUT 1.0
 
 // columns used for stat_info
-#define DBFS_STAT_COLS " inodes.type, inodes.mode, lo_size(data), count(*) "
-#define DBFS_STAT_COLS_NOAGGREGATE " inodes.type, inodes.mode, lo_size(data), NULL "
+#define DBFS_STAT_COLS " inodes.type, inodes.mode, dbfs_size(inodes.type, inodes.data, inodes.link_path), (SELECT COUNT(*) FROM inodes i LEFT JOIN file_tree ft ON (i.ino = ft.ino) WHERE i.ino = inodes.ino) AS nlink"
 
 /*
  * Convert the CHAR(4) inodes.type from SQL into a mode_t.

--- a/src/dbfs/dirop.c	Fri Oct 17 16:09:35 2008 +0300
+++ b/src/dbfs/dirop.c	Fri Oct 17 18:53:05 2008 +0300
@@ -94,8 +94,8 @@
     const char *sql =
         "SELECT"
         " file_tree.parent, inodes.type"
-        " FROM file_tree LEFT OUTER JOIN inodes ON (file_tree.inode = inodes.ino)"
-        " WHERE file_tree.inode = $1::int4";
+        " FROM file_tree LEFT OUTER JOIN inodes ON (file_tree.ino = inodes.ino)"
+        " WHERE file_tree.ino = $1::int4";
 
     static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {
         EVSQL_PARAM ( UINT32 ),
@@ -255,7 +255,7 @@
     const char *sql = 
         "SELECT"
         " file_tree.\"offset\", file_tree.name, inodes.ino, inodes.type"
-        " FROM file_tree LEFT OUTER JOIN inodes ON (file_tree.inode = inodes.ino)"
+        " FROM file_tree LEFT OUTER JOIN inodes ON (file_tree.ino = inodes.ino)"
         " WHERE file_tree.parent = $1::int4 AND file_tree.\"offset\" >= $2::int4"
         " LIMIT $3::int4";
 

--- a/src/dbfs/link.c	Fri Oct 17 16:09:35 2008 +0300
+++ b/src/dbfs/link.c	Fri Oct 17 18:53:05 2008 +0300
@@ -1,4 +1,9 @@
+#include <stdlib.h>
+#include <assert.h>
+#include <string.h>
+
 #include "dbfs.h"
+#include "trans.h"
 
 #include "../lib/log.h"
 
@@ -17,7 +22,7 @@
     if (0
         ||  evsql_result_uint32(res, 0, 0, &ino,        0 ) // inodes.ino
         ||  evsql_result_string(res, 0, 1, &type,       0 ) // inodes.type
-        ||  evsql_result_string(res, 0, 2, &link,       1 ) // inodes.symlink
+        ||  evsql_result_string(res, 0, 2, &link,       1 ) // inodes.link_path
     )
         EERROR(err = EIO, "invalid db data");
     
@@ -47,7 +52,7 @@
 
     const char *sql =
         "SELECT"
-        " inodes.ino, inodes.type, inodes.symlink"
+        " inodes.ino, inodes.type, inodes.link_path"
         " FROM inodes"
         " WHERE inodes.ino = $1::int4";
 
@@ -77,3 +82,184 @@
         EWARNING(err, "fuse_reply_err");
 
 }
+
+struct dbfs_symlink_ctx {
+    struct dbfs_trans base;
+
+    char *link, *name;
+    uint32_t ino, parent;
+};
+
+#define DBFS_SYMLINK_MODE 0777
+
+void dbfs_symlink_free (struct dbfs_trans *ctx_base) {
+    struct dbfs_symlink_ctx *ctx = (struct dbfs_symlink_ctx *) ctx_base;
+    
+    free(ctx->link);
+    free(ctx->name);
+}
+
+void dbfs_symlink_commit (struct dbfs_trans *ctx_base) {
+    struct dbfs_symlink_ctx *ctx = (struct dbfs_symlink_ctx *) ctx_base;
+    struct fuse_entry_param e;
+    int err;
+    
+    // build entry
+    e.ino = e.attr.st_ino = ctx->ino;
+    e.attr.st_mode = S_IFLNK | DBFS_SYMLINK_MODE;
+    e.attr.st_size = strlen(ctx->link);
+    e.attr.st_nlink = 1;
+    e.attr_timeout = e.entry_timeout = CACHE_TIMEOUT;
+
+    // reply
+    if ((err = fuse_reply_entry(ctx_base->req, &e)))
+        goto error;
+
+    // req good
+    ctx_base->req = NULL;
+    
+    // free
+    dbfs_trans_free(ctx_base);
+
+    // return
+    return;
+
+error:
+    dbfs_trans_fail(ctx_base, err);
+}
+
+void dbfs_symlink_filetree (const struct evsql_result_info *res, void *arg) {
+    struct dbfs_symlink_ctx *ctx = arg;
+    int err = EIO;
+    
+    // check results
+    if (_dbfs_check_res(res, 0, 0) < 0)
+        goto error;
+    
+    // commit
+    dbfs_trans_commit(&ctx->base);
+
+    // fallthrough for result_free
+    err = 0;
+
+error:
+    if (err)
+        dbfs_trans_fail(&ctx->base, err);
+
+    evsql_result_free(res);
+}
+
+void dbfs_symlink_inode (const struct evsql_result_info *res, void *arg) {
+    struct dbfs_symlink_ctx *ctx = arg;
+    struct dbfs *dbfs_ctx = fuse_req_userdata(ctx->base.req);
+    int err = EIO;
+    
+    // check result
+    if ((err = _dbfs_check_res(res, 1, 1)))
+        SERROR(err = err > 0 ? ENOENT : EIO);
+    
+    // get ino
+    if (evsql_result_uint32(res, 0, 0, &ctx->ino, 0))
+        goto error;
+
+    // insert file_tree entry
+    const char *sql = 
+        "INSERT"
+        " INTO file_tree (name, parent, ino)"
+        " VALUES ($1::varchar, $2::int4, $3::int4)";
+    
+    static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {
+        EVSQL_PARAM ( STRING ),
+        EVSQL_PARAM ( UINT32 ),
+        EVSQL_PARAM ( UINT32 ),
+
+        EVSQL_PARAMS_END
+    };
+
+    if (0
+        ||  evsql_param_string(&params, 0, ctx->name)
+        ||  evsql_param_uint32(&params, 1, ctx->parent)
+        ||  evsql_param_uint32(&params, 2, ctx->ino)
+    )
+        goto error;
+    
+    // query
+    if (evsql_query_params(dbfs_ctx->db, ctx->base.trans, sql, &params, dbfs_symlink_filetree, ctx))
+        goto error;
+    
+    // fallthrough for result_free
+    err = 0;
+
+error:
+    if (err)
+        dbfs_trans_fail(&ctx->base, err);
+
+    evsql_result_free(res);
+}
+
+void dbfs_symlink_begin (struct dbfs_trans *ctx_base) {
+    struct dbfs_symlink_ctx *ctx = (struct dbfs_symlink_ctx *) ctx_base;
+    struct dbfs *dbfs_ctx = fuse_req_userdata(ctx_base->req);
+
+    // insert inode
+    const char *sql = 
+        "INSERT"
+        " INTO inodes (type, mode, link_path)"
+        " VALUES ('LNK', $1::int2, $2::varchar)"
+        " RETURNING inodes.ino"; 
+    
+    static struct evsql_query_params params = EVSQL_PARAMS(EVSQL_FMT_BINARY) {
+        EVSQL_PARAM ( UINT16 ),
+        EVSQL_PARAM ( STRING ),
+
+        EVSQL_PARAMS_END
+    };
+
+    if (0
+        || evsql_param_uint16(&params, 0, DBFS_SYMLINK_MODE)
+        || evsql_param_string(&params, 1, ctx->link)
+    )
+        goto error;
+    
+    if (evsql_query_params(dbfs_ctx->db, ctx_base->trans, sql, &params, dbfs_symlink_inode, ctx) == NULL)
+        goto error;
+    
+    return;
+
+error:
+    dbfs_trans_fail(ctx_base, EIO);
+}
+
+void dbfs_symlink (struct fuse_req *req, const char *link, fuse_ino_t parent, const char *name) {
+    struct dbfs_symlink_ctx *ctx = NULL;
+
+    // alloc
+    if ((ctx = calloc(1, sizeof(*ctx))) == NULL)
+        ERROR("calloc");
+
+    // start trans
+    if (dbfs_trans_init(&ctx->base, req))
+        goto error;
+ 
+    // callbacks
+    ctx->base.free_fn = dbfs_symlink_free;
+    ctx->base.begin_fn = dbfs_symlink_begin;
+    ctx->base.commit_fn = dbfs_symlink_commit;
+   
+    // state
+    ctx->ino = 0;
+    ctx->parent = parent;
+    if (!((ctx->link = strdup(link)) && (ctx->name = strdup(name))))
+        ERROR("strdup");
+    
+    // log
+    INFO("[dbfs.symlink %p:%p] link=%s, parent=%lu, name=%s", ctx, req, link, parent, name);
+
+    // wait
+    return;
+
+error:
+    if (ctx)
+        dbfs_trans_fail(&ctx->base, EIO);
+}
+

--- a/src/dbfs/ops.h	Fri Oct 17 16:09:35 2008 +0300
+++ b/src/dbfs/ops.h	Fri Oct 17 18:53:05 2008 +0300
@@ -14,8 +14,9 @@
 void dbfs_getattr (struct fuse_req *req, fuse_ino_t ino, struct fuse_file_info *fi);
 void dbfs_setattr(struct fuse_req *req, fuse_ino_t ino, struct stat *attr, int to_set, struct fuse_file_info *fi);
 
-/* symlink.c */
+/* link.c */
 void dbfs_readlink (struct fuse_req *req, fuse_ino_t ino);
+void dbfs_symlink (struct fuse_req *req, const char *link, fuse_ino_t parent, const char *name);
 
 /* dirop.c */
 void dbfs_opendir (struct fuse_req *req, fuse_ino_t ino, struct fuse_file_info *fi);

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/dbfs/trans.c	Fri Oct 17 18:53:05 2008 +0300
@@ -0,0 +1,138 @@
+
+#include <stdlib.h>
+#include <assert.h>
+
+#include "trans.h"
+#include "../lib/log.h"
+
+void dbfs_trans_free (struct dbfs_trans *ctx) {
+    assert(!ctx->req);
+    assert(!ctx->trans);
+    
+    if (ctx->free_fn)
+        ctx->free_fn(ctx);
+
+    free(ctx);
+}
+
+void dbfs_trans_fail (struct dbfs_trans *ctx, int err) {
+    if (ctx->req) {
+        if ((err = fuse_reply_err(ctx->req, err)))
+            EWARNING(err, "fuse_reply_err: request hangs");
+
+        ctx->req = NULL;
+    }
+
+    if (ctx->trans) {
+        evsql_trans_abort(ctx->trans);
+
+        ctx->trans = NULL;
+    }
+
+    dbfs_trans_free(ctx);
+}
+
+static void dbfs_trans_error (struct evsql_trans *trans, void *arg) {
+    struct dbfs_trans *ctx = arg;
+
+    // deassociate trans
+    ctx->trans = NULL;
+
+    // log error
+    INFO("\t[dbfs_trans.err %p:%p] %s", ctx, ctx->req, evsql_trans_error(trans));
+
+    // mark
+    if (ctx->err_ptr)
+        *ctx->err_ptr = EIO;
+    
+    // fail
+    dbfs_trans_fail(ctx, EIO);
+}
+
+static void dbfs_trans_ready (struct evsql_trans *trans, void *arg) {
+    struct dbfs_trans *ctx = arg;
+
+    // associate trans
+    ctx->trans = trans;
+
+    // log
+    INFO("\t[dbfs_trans.ready %p:%p] -> trans=%p", ctx, ctx->req, trans);
+
+    // trigger the callback
+    ctx->begin_fn(ctx);
+}
+
+static void dbfs_trans_done (struct evsql_trans *trans, void *arg) {
+    struct dbfs_trans *ctx = arg;
+
+    // deassociate trans
+    ctx->trans = NULL;
+
+    // log
+    INFO("\t[dbfs_trans.done %p:%p]", ctx, ctx->req);
+
+    // trigger the callback
+    ctx->commit_fn(ctx);
+}
+
+int dbfs_trans_init (struct dbfs_trans *ctx, struct fuse_req *req) {
+    struct dbfs *dbfs_ctx = fuse_req_userdata(req);
+    int err;
+
+    // store
+    ctx->req = req;
+
+    // trans
+    if ((ctx->trans = evsql_trans(dbfs_ctx->db, EVSQL_TRANS_SERIALIZABLE, dbfs_trans_error, dbfs_trans_ready, dbfs_trans_done, ctx)) == NULL)
+        EERROR(err = EIO, "evsql_trans");
+
+    // good
+    return 0;
+
+error:
+    return -1;
+}
+
+struct dbfs_trans *dbfs_trans_new (struct fuse_req *req) {
+    struct dbfs_trans *ctx = NULL;
+
+    // alloc
+    if ((ctx = calloc(1, sizeof(*ctx))) == NULL)
+        ERROR("calloc");
+
+    // init
+    if (dbfs_trans_init(ctx, req))
+        goto error;
+
+    // good
+    return ctx;
+    
+error:
+    free(ctx);
+
+    return NULL;
+}
+
+void dbfs_trans_commit (struct dbfs_trans *ctx) {
+    int err, trans_err = 0;
+
+    // detect errors
+    ctx->err_ptr = &trans_err;
+    
+    // attempt commit
+    if (evsql_trans_commit(ctx->trans))
+        SERROR(err = EIO);
+    
+    // drop err_ptr
+    ctx->err_ptr = NULL;
+
+    // ok, wait for done or error
+    return;
+
+error:
+    // fail if not already failed
+    if (!trans_err)
+        dbfs_trans_fail(ctx, err);
+}
+
+

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/dbfs/trans.h	Fri Oct 17 18:53:05 2008 +0300
@@ -0,0 +1,69 @@
+#ifndef DBFS_TRANS_H
+#define DBFS_TRANS_H
+
+/*
+ * Support for single-fuse_req transactions.
+ */
+
+#include "dbfs.h"
+
+// forward-declaration
+struct dbfs_trans;
+
+// generic callback
+typedef void (*dbfs_trans_cb) (struct dbfs_trans *ctx);
+
+/*
+ * Request/transaction state.
+ */
+struct dbfs_trans {
+    struct fuse_req *req;
+    struct evsql_trans *trans;
+    
+    // called when the dbfs_trans is being free'd
+    dbfs_trans_cb free_fn;
+
+    // called once the transaction is ready
+    dbfs_trans_cb begin_fn;
+
+    // called once the transaction has been commited
+    dbfs_trans_cb commit_fn;
+
+    // set by trans_error to EIO if !NULL
+    int *err_ptr;
+};
+
+/*
+ * Call from commit_fn once you've set req to NULL. Also called from trans_fail.
+ *
+ * Will call free_fn if present.
+ */
+void dbfs_trans_free (struct dbfs_trans *ctx);
+
+/*
+ * Fail the dbfs_trans, aborting any trans, erroring out any req and freeing the ctx.
+ */
+void dbfs_trans_fail (struct dbfs_trans *ctx, int err);
+
+/*
+ * Initialize the ctx with the given req (stays the same during the lifetime), and opens the transaction.
+ *
+ * You should set the callback functions after/before calling this.
+ *
+ * begin_fn will be called once the transaction is open, if that fails, the req will be errored for you.
+ *
+ * If opening the transaction fails, this will return nonzero and not touch req, otherwise zero.
+ */
+int dbfs_trans_init (struct dbfs_trans *ctx, struct fuse_req *req);
+
+/*
+ * Same as init, but allocates/frees-on-error the dbfs_trans for you.
+ */
+struct dbfs_trans *dbfs_trans_new (struct fuse_req *req);
+
+/*
+ * Commit the dbfs_trans. After calling this, the ctx may or may not be valid, and commit_fn may or may not be called.
+ */
+void dbfs_trans_commit (struct dbfs_trans *ctx);
+
+#endif /* DBFS_TRANS_H */

--- a/src/lib/error.h	Fri Oct 17 16:09:35 2008 +0300
+++ b/src/lib/error.h	Fri Oct 17 18:53:05 2008 +0300
@@ -7,6 +7,7 @@
 #define PERROR(...) do { perr_func(__func__, __VA_ARGS__); goto error; } while (0)
 #define EERROR(_err, ...) do { eerr_func(__func__, (_err), __VA_ARGS__); goto error; } while (0)
 #define NERROR(...) do { err_func_nonl(__func__, __VA_ARGS__); goto error; } while (0)
+#define SERROR(_err) do { (_err); goto error; } while (0)
 
 // XXX: replace with *err_func(...) + exit(EXIT_FAILURE)
 #define FATAL(...) err_func_exit(__func__, __VA_ARGS__)

--- a/src/lib/log.c	Fri Oct 17 16:09:35 2008 +0300
+++ b/src/lib/log.c	Fri Oct 17 18:53:05 2008 +0300
@@ -3,12 +3,16 @@
 #include <stdarg.h>
 #include <string.h>
 #include <errno.h>
+#include <assert.h>
 
 #include "log.h"
 
 static void _generic_err_vargs (int flags, const char *func, int err, const char *fmt, va_list va) {
     FILE *stream = flags & LOG_DISPLAY_STDERR ? stderr : stdout;
 
+    if (!fmt)
+        return;
+
     if (flags & LOG_DISPLAY_FATAL)
         fprintf(stream, "FATAL: ");
 
@@ -35,6 +39,8 @@
 void _generic_err_exit (int flags, const char *func, int err, const char *fmt, ...) {
     va_list va;
 
+    assert(fmt);
+
     va_start(va, fmt);
     _generic_err_vargs(flags | LOG_DISPLAY_FATAL, func, err, fmt, va);
     va_end(va);