# Mdbfs was written my Matt Heath [email protected] in July 2003
# It was heavily adapted from 
# /appl/cmd/dbfs.b

# which contains this :

# Copyright � 1999 Vita Nuova Limited.  All rights reserved.
# Revisions copyright � 2002 Vita Nuova Holdings Limited.  All rights reserved.

# I'm not sure what I'm supposed to put here. The Licence says that if I supply a binary I agree to supply the source
# and any derivatative works (i.e. this one) carry the same burden
# /me shrugs and carries on

# on with the show :
# Mdbfs is an extension to dbfs that takes it into the second dimension with both rows & columns
# it mounts the data read only whereas dbfs was read/write.
# Based on my tabfs the idea was that it would consume less memory but with the primary key hashtable 
# it actually uses more !
# mind you the trade off is time, it took six hours to run out of 256Mb of memory with the last one 
# at least this one has the grace to do it in 30 seconds

# Usage: mdbfs [-a|-b] [-D] [-1] [-p n] [-P] file mountpoint

# -a|-b	- after / before
# -D	- print each Tmsg/Rmsg 
# -1    - use the first line of the file as the column names (otherwise they are numbered)
# -p n  - use column number n as the primary key and therefore the name of the directory for each row (otherwise numbered)
#         if this column contains duplicates they will still be created but with distinct Qids but the Hashtable will get screwed
#         so it's best not to do that, I've not played with that so be it on your own head
# -P	- print every cell of the database in qid order, one per line, data is bracketed thus : >%s<
#           I was using it for debugging but left it in
# file	- The file is split into records on each non-empty line using string->unquoted. The number of columns is calculated
#	  from the first line. If a row contains less they are still included, any extra are discarded

implement Mdbfs;

include "sys.m";
	sys: Sys;
	Qid: import Sys;

include "draw.m";

include "arg.m";

include "styx.m";
	styx: Styx;
	Tmsg, Rmsg: import styx;

include "string.m";
	str: String;

include "styxservers.m";
	styxservers: Styxservers;
	Fid, Styxserver, Navigator, Navop: import styxservers;
	Enotfound, Eperm, Ebadarg: import styxservers;

include "bufio.m";
	bufio: Bufio;
	Iobuf: import bufio;

include "hash.m";
	hash : Hash;
	HashTable, HashVal : import hash;

Record: adt {
	id:		int;		# file number in directory
	dirty:	int;		# modified but not written
	vers:		int;		# version
	data:		string;

	print:	fn(r: self ref Record);
	qid:	fn (r: self ref Record): Sys->Qid;
};

Database: adt {
	name:	string;
	file:	ref Iobuf;
	records:	array of ref Record;
	dirty:	int;
	vers:		int;
	row_count :	int;
	first_row_contains_column_names : int;
	primary_key_column_index : int;

	column_names_string : string;
	column_names_hash : ref HashTable;
	column_names_array : array of string;
	column_count : int;

	pk_hash : ref HashTable;

	getrec: fn(db: self ref Database): (string, array of string, string);
	findrec:	fn(db: self ref Database, row, column: int): ref Record;
	set_column_names: fn(db: self ref Database, names : string);
	set_column_names_to_numbers: fn(db: self ref Database, num : int);
	print: fn(db : self ref Database);
};

Mdbfs: module
{
	init: fn(nil: ref Draw->Context, nil: list of string);
};

Qroot, Qrow, Qcolumn: con iota;

stderr: ref Sys->FD;
database: ref Database;
user: string;
Eremoved: con "file removed";
progname : con "2dbfs";

rootfiles := array[] of {"/", "data", "src_filename", "column_names"};

usage()
{
	sys->fprint(stderr, "Usage: %s [-a|-b] [-D] [-1] [-p n] [-P] file mountpoint\n", progname);
	raise "fail:usage";
}

nomod(s: string)
{
	sys->fprint(stderr, "%s: can't load %s: %r\n", progname, s);
	raise "fail:load";
}

nomem(s : string)
{
	sys->fprint(stderr, "%s : can't allocate memory - %s", progname, s);
	raise "fail:memory";
}

init(nil: ref Draw->Context, args: list of string)
{
	sys = load Sys Sys->PATH;
	sys->pctl(Sys->FORKFD|Sys->NEWPGRP, nil);

	stderr = sys->fildes(2);

	styx = load Styx Styx->PATH;
	if(styx == nil)
		nomod(Styx->PATH);
	styx->init();
	styxservers = load Styxservers Styxservers->PATH;
	if(styxservers == nil)
		nomod(Styxservers->PATH);
	styxservers->init(styx);

	bufio = load Bufio Bufio->PATH;
	if(bufio == nil)
		nomod(Bufio->PATH);

	str = load String String->PATH;
	if(str == nil)
		nomod(String->PATH);

	hash = load Hash Hash->PATH;
	if(hash == nil)
		nomod(Hash->PATH);

	arg := load Arg Arg->PATH;
	if(arg == nil)
		nomod(Arg->PATH);
	arg->init(args);
	flags := Sys->MREPL;
	copt := 0;
	first_row_contains_column_names := 0;
	primary_key_column_index := -1;
	print_db := 0;

	while((o := arg->opt()) != 0)
		case o {
		'a' =>	flags = Sys->MAFTER;
		'b' =>	flags = Sys->MBEFORE;
		'D' =>	styxservers->traceset(1);
		'P' =>	print_db = 1;
		'1' =>	first_row_contains_column_names = 1;
		'p' =>	primary_key_column_index = int arg->arg();
		* =>		usage();
		}
	args = arg->argv();
	arg = nil;

	if(len args != 2)
		usage();

	file := hd args;
	args = tl args;
	mountpt := hd args;

	df := bufio->open(file, Sys->OREAD);

	if(df == nil){
		sys->fprint(stderr, "%s: can't open %s: %r\n", progname, file);
		raise "fail:open";
	}
	(db, err) := dbread(ref Database(file, df, nil, 0, 0, 0, first_row_contains_column_names, primary_key_column_index, nil, nil, nil, 0, nil));
	if(db == nil){
		sys->fprint(stderr, "%s: can't read %s: %s\n", progname, file, err);
		raise "fail:dbread";
	}
	db.file = nil;

	database = db;

	sys->pctl(Sys->FORKFD, nil);

	user = rf("/dev/user");
	if(user == nil)
		user = "inferno";

	fds := array[2] of ref Sys->FD;
	if(sys->pipe(fds) < 0){
		sys->fprint(stderr, "%s: can't create pipe: %r\n", progname);
		raise "fail:pipe";
	}

	navops := chan of ref Navop;
	spawn navigator(navops);

	(tchan, srv) := Styxserver.new(fds[0], Navigator.new(navops), big Qroot);
	fds[0] = nil;

	pidc := chan of int;
	spawn serveloop(tchan, srv, pidc, navops);
	<-pidc;

	if (print_db)
		db.print();

	if(sys->mount(fds[1], nil, mountpt, flags, nil) < 0) {
		sys->fprint(stderr, "%s: mount failed: %r\n", progname);
		raise "fail:mount";
	}
}

rf(f: string): string
{
	fd := sys->open(f, Sys->OREAD);
	if(fd == nil)
		return nil;
	b := array[Sys->NAMEMAX] of byte;
	n := sys->read(fd, b, len b);
	if(n < 0)
		return nil;
	return string b[0:n];
}

dbread(db: ref Database): (ref Database, string)
{
	db.file.seek(big 0, Sys->SEEKSTART);
	rl: list of array of string;

	(data, r, err) := db.getrec();

	if(err != nil)
		return (nil, err);	
	if(db.first_row_contains_column_names == 1) {
		db.set_column_names(data);
		(nil, r, err) = db.getrec();
		if(err != nil)
			return (nil, err);	
	} else 
		db.set_column_names_to_numbers(len r);

	rl = r :: rl;
	db.row_count = 1;

	if(db.column_count  < 1) 
		return (nil, "need at least one column name");
	
	for(;;){
		(nil, r, err) = db.getrec();
		if(err != nil)
			return (nil, err);	
		if(r == nil)
			break;
		rl = r :: rl;
		db.row_count++;
	}

	record_count := db.row_count * db.column_count;
	db.records = array[record_count] of ref Record;
	db.pk_hash = hash->new(record_count);

	row_num := db.row_count -1;
	pk_value : string;

	for(; rl != nil; rl = tl rl) {
		r = hd rl;

		if(db.primary_key_column_index < 0)
			pk_value = sys->sprint("%d", row_num);
		else
			if (len r < db.column_count)
				pk_value = sys->sprint("!%d!", row_num);
			else
				pk_value = r[db.primary_key_column_index];

		db.pk_hash.insert(pk_value, HashVal(row_num--, 0.0, nil));

		for (column_count := db.column_count - 1; column_count  >= 0; column_count--) {
			rec :=  ref Record(--record_count, 0, 0, r[column_count]);
			db.records[record_count] = rec;
		}
	}

	return (db, nil);
}



Database.getrec(db: self ref Database): (string, array of string, string)
{
	data := db.file.gets('\n');
	if(data == nil)
		return (nil, nil, nil);		# BUG: distinguish i/o error from EOF? - nah 

	(data, nil) = str->splitl(data, "\n");

	if(data == "")
		return db.getrec(); # skip blank lines

	return (data, list_to_array(str->unquoted(data)), nil);
}


Database.findrec(db: self ref Database, row, column: int): ref Record
{
	a := db.records[row].data[column];
	return nil;
}

Database.set_column_names(db: self ref Database, names : string)
{
	uq := str->unquoted(names);
	db.column_names_hash  = list_to_hash(uq);
	db.column_names_array  = list_to_array(uq);
	db.column_names_string = names;
	db.column_count = len db.column_names_array;
}

Database.set_column_names_to_numbers(db: self ref Database, num : int)
{
	namestring := sys->sprint("%d", 0);
	for(i:=1; i < num; i++)
		namestring += sys->sprint(" %d", i);
	db.set_column_names(namestring);
}

Database.print(db: self ref Database) 
{
	record_count := len db.records;

	for( i:=0; i < record_count; i++)
		db.records[i].print();
}

Record.qid(r: self ref Record): Sys->Qid
{
	return Sys->Qid(QPATH(r.id, Qcolumn), r.vers, Sys->QTFILE);
}

Record.print(r: self ref Record)
{
	sys->fprint(stderr, "id: %6d	qid: %16.16bx	data: >%s<\n", r.id, r.qid().path, r.data);
}

serveloop(tchan: chan of ref Tmsg, srv: ref Styxserver, pidc: chan of int, navops: chan of ref Navop)
{
	pidc <-= sys->pctl(Sys->FORKNS|Sys->NEWFD, 1::2::srv.fd.fd::nil);
Serve:
	while((gm := <-tchan) != nil){
		pick m := gm {
		Readerror =>
			sys->fprint(stderr, "%s: fatal read error: %s\n", progname, m.error);
			break Serve;

		Read =>
			(c, err) := srv.canread(m);
			if(c == nil){
				srv.reply(ref Rmsg.Error(m.tag, err));
				break;
			}
			if(c.qtype & Sys->QTDIR){

				srv.read(m);	# does readdir
				break;
			}

			case TYPE(c.path) {
			Qroot =>
				case FILENO(c.path) {
					2 => srv.reply(styxservers->readstr(m, database.name));
					3 => srv.reply(styxservers->readstr(m, database.column_names_string));
					* => srv.default(gm);
				}
			Qcolumn =>
				rec_index := FILENO(c.path);

				if (rec_index < len database.records)
					srv.reply(styxservers->readstr(m, database.records[rec_index].data));
				else
					srv.default(gm);
			* => 
				srv.default(gm);
			}

		Write =>
		# nothing writable atm.
			(c, merr) := srv.canwrite(m);
			if(c == nil){
				srv.reply(ref Rmsg.Error(m.tag, merr));
				break;
			}
			case TYPE(c.path) {
			Qroot =>
				case FILENO(c.path) {
				#	2 => srv.reply(styxservers->readstr(m, database.name));
				#	3 => srv.reply(styxservers->readstr(m, database.column_names_string));
				* => 
					srv.default(gm);
				}
			}

		Clunk =>
			srv.clunk(m);
		* =>
			srv.default(gm);
		}	
	}
	navops <-= nil;		# shut down navigator
}

dir(qid: Sys->Qid, name: string, length: big, uid: string, perm: int): ref Sys->Dir
{
	d := ref sys->zerodir;
	d.qid = qid;
	if(qid.qtype & Sys->QTDIR)
		perm |= Sys->DMDIR;
	d.mode = perm;
	d.name = name;
	d.uid = uid;
	d.gid = uid;
	d.length = length;
	return d;
}

dirgen(p: big): (ref Sys->Dir, string)
{
	file_index := FILENO(p) ;
	case TYPE(p) {
	Qroot =>
		case file_index {
		0 => 
			qid := Qid(QPATH(file_index, Qroot), database.vers, Sys->QTDIR);
			return (dir(qid, "/", big 0, user, 8r700), nil);
		1 =>
			qid := Qid(QPATH(file_index, Qroot), database.vers, Sys->QTDIR);
			return (dir(qid, "data", big 0, user, 8r555), nil);
		* =>
			if (file_index < len rootfiles) {
				qid := Qid(QPATH(file_index, Qroot), database.vers, Sys->QTFILE);
				return (dir(qid, rootfiles[file_index], big 0, user, 8r666), nil);
			}
		}
	Qrow =>
		pk_value : string;

		if (database.primary_key_column_index < 0) {
			pk_value = sys->sprint("%d", file_index);
		} else {
			cell := file_index * database.column_count + database.primary_key_column_index;

			if (cell < len database.records)
				pk_value = database.records[cell].data;
		}

		if (pk_value != nil) {
			qid := Qid(QPATH(file_index, Qrow), database.vers, Sys->QTDIR);
			return (dir(qid, pk_value, big 0, user, 8r555), nil);
		}

	Qcolumn =>
		column_number := file_index % database.column_count;

		if (file_index < len database.records) 
			return (dir(database.records[file_index].qid(), database.column_names_array[column_number], big 0, user, 8r444), nil);
	}

	return (nil, Enotfound);
}

navigator(navops: chan of ref Navop)
{
	while((m := <-navops) != nil){
		pick n := m {
		Stat =>
			n.reply <-= dirgen(n.path);
		Walk =>
			case (TYPE(n.path)) {
			Qroot => 
				case FILENO(n.path) {
				0 =>
					case n.name {
						"data" => n.reply <-= dirgen(QPATH(1, Qroot));
						"src_filename" => n.reply <-= dirgen(QPATH(2, Qroot));
						"column_names" => n.reply <-= dirgen(QPATH(3, Qroot));
						* => 	n.reply <-= (nil, Enotfound);
					}
				1 =>
					pk := database.pk_hash.find(n.name);

					if(pk == nil)  {
						n.reply <-= (nil, Enotfound);
						continue;
					}

					n.path = QPATH(pk.i, Qrow);
					n.reply <-= dirgen(n.path);
				* =>
					n.reply <-= (nil, "not a directory");
				}
			Qrow =>
				row_index := FILENO(n.path);
				column_hashval := database.column_names_hash.find(n.name);
				if(column_hashval == nil) {
					n.reply <-= (nil, Enotfound);
					continue;
				}
				n.path = QPATH(column_hashval.i + database.column_count * row_index, Qcolumn);
				n.reply <-= dirgen(n.path);
			* => 
				n.reply <-= (nil, "not a directory");
			}

		Readdir =>
			case (TYPE(n.path)) {
			Qroot =>
				case FILENO(n.path) {
				0 =>
					for(i := n.offset; --n.count >= 0 && i < len rootfiles; i++)
						n.reply <-= dirgen(QPATH(i,Qroot));	
				1 =>
					for(i := n.offset; --n.count >= 0 && i < database.row_count ; i++) 
						n.reply <-= dirgen(QPATH(i,Qrow));
				}
				n.reply <-= (nil, nil);
			Qrow =>
				i_start := FILENO(n.path) * database.column_count;

				for(i := n.offset; --n.count >= 0 && i < database.column_count ; i++) 
					n.reply <-= dirgen(QPATH(i_start++, Qcolumn));

				n.reply <-= (nil, nil);
			* =>
				n.reply <-= (nil, "not a directory");
			}
		}
	}
}

QPATH(index, qtype: int): big
{
	bigi := big index;
	bigi = bigi << 32;

	return bigi | big qtype;
}

TYPE(path: big): int
{
	return int path;
}

FILENO(path: big) : int
{
	return int (path >> 32) ;
}

list_to_array(lizt : list of string) : array of string 
{
	list_length := len lizt;
	hooray := array[list_length] of string;
	for(i := 0;  i < list_length ; i++) {
		hooray[i] = hd lizt;
		lizt = tl lizt;
	}

	return hooray;
}

list_to_hash(lizt : list of string) : ref HashTable 
{
	list_length := len lizt;

	hsh := hash->new(list_length);

	j := 0;

	for(i := list_length - 1;  i > -1 ; i--) {
		v := HashVal(j++, 0.0, nil);
		hsh.insert(hd lizt, v);
		lizt = tl lizt;
	}

	return hsh;
}