Complete first alpha version

api/config.go

@@ -0,0 +1,49 @@
+package api
+
+import (
+	"flag"
+
+	"github.com/Safing/portbase/config"
+	"github.com/Safing/portbase/log"
+)
+
+var (
+	listenAddressFlag   string
+	listenAddressConfig config.StringOption
+)
+
+func init() {
+	flag.StringVar(&listenAddressFlag, "api-address", "", "override api listen address")
+}
+
+func checkFlags() error {
+	if listenAddressFlag != "" {
+		log.Warning("api: api/listenAddress config is being overridden by -api-address flag")
+	}
+	return nil
+}
+
+func getListenAddress() string {
+	if listenAddressFlag != "" {
+		return listenAddressFlag
+	}
+	return listenAddressConfig()
+}
+
+func registerConfig() error {
+	err := config.Register(&config.Option{
+		Name:            "API Address",
+		Key:             "api/listenAddress",
+		Description:     "Define on what IP and port the API should listen on. Be careful, changing this may become a security issue.",
+		ExpertiseLevel:  config.ExpertiseLevelExpert,
+		OptType:         config.OptTypeString,
+		DefaultValue:    "127.0.0.1:18",
+		ValidationRegex: "^([0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}:[0-9]{1,5}|\\[[:0-9A-Fa-f]+\\]:[0-9]{1,5})$",
+	})
+	if err != nil {
+		return err
+	}
+	listenAddressConfig = config.GetAsString("api/listenAddress", "127.0.0.1:18")
+
+	return nil
+}

api/database.go

@@ -27,12 +27,12 @@ const (
 	dbMsgTypeDel     = "del"
 	dbMsgTypeWarning = "warning"
 
-	dbApiSeperator = "|"
+	dbAPISeperator = "|"
 	emptyString    = ""
 )
 
 var (
-	dbApiSeperatorBytes = []byte(dbApiSeperator)
+	dbAPISeperatorBytes = []byte(dbAPISeperator)
 )
 
 // DatabaseAPI is a database API instance.
@@ -76,6 +76,8 @@ func startDatabaseAPI(w http.ResponseWriter, r *http.Request) {
 
 	go new.handler()
 	go new.writer()
+
+	log.Infof("api request: init websocket %s %s", r.RemoteAddr, r.RequestURI)
 }
 
 func (api *DatabaseAPI) handler() {
@@ -210,16 +212,16 @@ func (api *DatabaseAPI) writer() {
 
 func (api *DatabaseAPI) send(opID []byte, msgType string, msgOrKey string, data []byte) {
 	c := container.New(opID)
-	c.Append(dbApiSeperatorBytes)
+	c.Append(dbAPISeperatorBytes)
 	c.Append([]byte(msgType))
 
 	if msgOrKey != emptyString {
-		c.Append(dbApiSeperatorBytes)
+		c.Append(dbAPISeperatorBytes)
 		c.Append([]byte(msgOrKey))
 	}
 
 	if len(data) > 0 {
-		c.Append(dbApiSeperatorBytes)
+		c.Append(dbAPISeperatorBytes)
 		c.Append(data)
 	}
 
@@ -270,14 +272,16 @@ func (api *DatabaseAPI) processQuery(opID []byte, q *query.Query) (ok bool) {
 	}
 
 	for r := range it.Next {
+		r.Lock()
 		data, err := r.Marshal(r, record.JSON)
+		r.Unlock()
 		if err != nil {
 			api.send(opID, dbMsgTypeWarning, err.Error(), nil)
 		}
 		api.send(opID, dbMsgTypeOk, r.Key(), data)
 	}
-	if it.Err != nil {
+	if it.Err() != nil {
-		api.send(opID, dbMsgTypeError, it.Err.Error(), nil)
+		api.send(opID, dbMsgTypeError, it.Err().Error(), nil)
 		return false
 	}
 
@@ -320,7 +324,9 @@ func (api *DatabaseAPI) registerSub(opID []byte, q *query.Query) (sub *database.
 
 func (api *DatabaseAPI) processSub(opID []byte, sub *database.Subscription) {
 	for r := range sub.Feed {
+		r.Lock()
 		data, err := r.Marshal(r, record.JSON)
+		r.Unlock()
 		if err != nil {
 			api.send(opID, dbMsgTypeWarning, err.Error(), nil)
 		}

api/enriched-response.go

@@ -4,11 +4,13 @@ import (
 	"net/http"
 )
 
+// EnrichedResponseWriter is a wrapper for http.ResponseWriter for better information extraction.
 type EnrichedResponseWriter struct {
 	http.ResponseWriter
 	Status int
 }
 
+// NewEnrichedResponseWriter wraps a http.ResponseWriter.
 func NewEnrichedResponseWriter(w http.ResponseWriter) *EnrichedResponseWriter {
 	return &EnrichedResponseWriter{
 		w,
@@ -16,6 +18,7 @@ func NewEnrichedResponseWriter(w http.ResponseWriter) *EnrichedResponseWriter {
 	}
 }
 
+// WriteHeader wraps the original WriteHeader method to extract information.
 func (ew *EnrichedResponseWriter) WriteHeader(code int) {
 	ew.Status = code
 	ew.ResponseWriter.WriteHeader(code)

api/main.go

@@ -5,18 +5,18 @@ import (
 )
 
 func init() {
-	modules.Register("api", prep, start, stop, "database")
+	modules.Register("api", prep, start, nil, "database")
 }
 
 func prep() error {
-	return nil
+	err := checkFlags()
+	if err != nil {
+		return err
+	}
+	return registerConfig()
 }
 
 func start() error {
 	go Serve()
 	return nil
 }
-
-func stop() error {
-	return nil
-}

api/router.go

@@ -12,6 +12,7 @@ var (
 	additionalRoutes map[string]http.Handler
 )
 
+// RegisterAdditionalRoute registers an additional route with the API endoint.
 func RegisterAdditionalRoute(path string, handler http.Handler) {
 	if additionalRoutes == nil {
 		additionalRoutes = make(map[string]http.Handler)
@@ -19,6 +20,7 @@ func RegisterAdditionalRoute(path string, handler http.Handler) {
 	additionalRoutes[path] = handler
 }
 
+// RequestLogger is a logging middleware
 func RequestLogger(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ew := NewEnrichedResponseWriter(w)
@@ -27,6 +29,7 @@ func RequestLogger(next http.Handler) http.Handler {
 	})
 }
 
+// Serve starts serving the API endpoint.
 func Serve() {
 
 	router := mux.NewRouter()
@@ -41,7 +44,7 @@ func Serve() {
 	http.Handle("/", router)
 	http.HandleFunc("/api/database/v1", startDatabaseAPI)
 
-	address := "127.0.0.1:18"
+	address := getListenAddress()
 	log.Infof("api: starting to listen on %s", address)
 	log.Errorf("api: failed to listen on %s: %s", address, http.ListenAndServe(address, nil))
 }

build

@@ -49,4 +49,4 @@ echo "Run the compiled binary with the -version flag to see the information incl
 
 # build
 BUILD_PATH="github.com/Safing/portbase/info"
-go build -ldflags "-X ${BUILD_PATH}.commit=${BUILD_COMMIT} -X ${BUILD_PATH}.buildOptions=${BUILD_BUILDOPTIONS} -X ${BUILD_PATH}.buildUser=${BUILD_USER} -X ${BUILD_PATH}.buildHost=${BUILD_HOST} -X ${BUILD_PATH}.buildDate=${BUILD_DATE} -X ${BUILD_PATH}.buildSource=${BUILD_SOURCE}" $*
+go build -ldflags "-X ${BUILD_PATH}.commit=${BUILD_COMMIT} -X ${BUILD_PATH}.buildOptions=${BUILD_BUILDOPTIONS} -X ${BUILD_PATH}.buildUser=${BUILD_USER} -X ${BUILD_PATH}.buildHost=${BUILD_HOST} -X ${BUILD_PATH}.buildDate=${BUILD_DATE} -X ${BUILD_PATH}.buildSource=${BUILD_SOURCE}" $@

database/controller.go

@@ -19,7 +19,12 @@ type Controller struct {
 	subscriptions []*Subscription
 
 	writeLock sync.RWMutex
+	//  Lock: nobody may write
+	// RLock: concurrent writing
 	readLock sync.RWMutex
+	//  Lock: nobody may read
+	// RLock: concurrent reading
+
 	migrating   *abool.AtomicBool // TODO
 	hibernating *abool.AtomicBool // TODO
 }
@@ -101,9 +106,6 @@ func (c *Controller) Put(r record.Record) (err error) {
 		return ErrReadOnly
 	}
 
-	r.Lock()
-	defer r.Unlock()
-
 	// process hooks
 	for _, hook := range c.hooks {
 		if hook.h.UsesPrePut() && hook.q.Matches(r) {
@@ -160,6 +162,9 @@ func (c *Controller) Query(q *query.Query, local, internal bool) (*iterator.Iter
 // PushUpdate pushes a record update to subscribers.
 func (c *Controller) PushUpdate(r record.Record) {
 	if c != nil {
+		c.readLock.RLock()
+		defer c.readLock.RUnlock()
+
 		for _, sub := range c.subscriptions {
 			if r.Meta().CheckPermission(sub.local, sub.internal) && sub.q.Matches(r) {
 				select {
@@ -171,19 +176,28 @@ func (c *Controller) PushUpdate(r record.Record) {
 	}
 }
 
+func (c *Controller) addSubscription(sub *Subscription) {
+	c.readLock.Lock()
+	defer c.readLock.Unlock()
+	c.writeLock.Lock()
+	defer c.writeLock.Unlock()
+
+	c.subscriptions = append(c.subscriptions, sub)
+}
+
 func (c *Controller) readUnlockerAfterQuery(it *iterator.Iterator) {
 	<-it.Done
 	c.readLock.RUnlock()
 }
 
-// Maintain runs the Maintain method no the storage.
+// Maintain runs the Maintain method on the storage.
 func (c *Controller) Maintain() error {
 	c.writeLock.RLock()
 	defer c.writeLock.RUnlock()
 	return c.storage.Maintain()
 }
 
-// MaintainThorough runs the MaintainThorough method no the storage.
+// MaintainThorough runs the MaintainThorough method on the storage.
 func (c *Controller) MaintainThorough() error {
 	c.writeLock.RLock()
 	defer c.writeLock.RUnlock()

database/database_test.go

@@ -98,8 +98,8 @@ func testDatabase(t *testing.T, storageType string) {
 	for _ = range it.Next {
 		cnt++
 	}
-	if it.Err != nil {
+	if it.Err() != nil {
-		t.Fatal(it.Err)
+		t.Fatal(it.Err())
 	}
 	if cnt != 2 {
 		t.Fatal("expected two records")

database/dbmodule/db.go

@@ -31,7 +31,7 @@ func prep() error {
 func start() error {
 	err := database.Initialize(databaseDir)
 	if err == nil {
-		go maintainer()
+		startMaintainer()
 	}
 	return err
 }

database/dbmodule/maintenance.go

@@ -7,10 +7,19 @@ import (
 	"github.com/Safing/portbase/log"
 )
 
-func maintainer() {
+var (
-  ticker := time.NewTicker(10 * time.Minute)
+	maintenanceShortTickDuration = 10 * time.Minute
-  longTicker := time.NewTicker(1 * time.Hour)
+	maintenanceLongTickDuration  = 1 * time.Hour
+)
+
+func startMaintainer() {
 	maintenanceWg.Add(1)
+	go maintenanceWorker()
+}
+
+func maintenanceWorker() {
+	ticker := time.NewTicker(maintenanceShortTickDuration)
+	longTicker := time.NewTicker(maintenanceLongTickDuration)
 
 	for {
 		select {

database/interface.go

@@ -178,6 +178,9 @@ func (i *Interface) Put(r record.Record) error {
 		return err
 	}
 
+	r.Lock()
+	defer r.Unlock()
+
 	i.options.Apply(r)
 
 	i.updateCache(r)
@@ -191,6 +194,12 @@ func (i *Interface) PutNew(r record.Record) error {
 		return err
 	}
 
+	r.Lock()
+	defer r.Unlock()
+
+	if r.Meta() == nil {
+		r.SetMeta(&record.Meta{})
+	}
 	r.Meta().Reset()
 	i.options.Apply(r)
 	i.updateCache(r)
@@ -296,17 +305,12 @@ func (i *Interface) Subscribe(q *query.Query) (*Subscription, error) {
 		return nil, err
 	}
 
-	c.readLock.Lock()
-	defer c.readLock.Unlock()
-	c.writeLock.Lock()
-	defer c.writeLock.Unlock()
-
 	sub := &Subscription{
 		q:        q,
 		local:    i.options.Local,
 		internal: i.options.Internal,
 		Feed:     make(chan record.Record, 100),
 	}
-	c.subscriptions = append(c.subscriptions, sub)
+	c.addSubscription(sub)
 	return sub, nil
 }

database/iterator/iterator.go

@@ -1,6 +1,10 @@
 package iterator
 
 import (
+	"sync"
+
+	"github.com/tevino/abool"
+
 	"github.com/Safing/portbase/database/record"
 )
 
@@ -8,7 +12,10 @@ import (
 type Iterator struct {
 	Next chan record.Record
 	Done chan struct{}
-	Err  error
+
+	errLock    sync.Mutex
+	err        error
+	doneClosed *abool.AtomicBool
 }
 
 // New creates a new Iterator.
@@ -16,11 +23,32 @@ func New() *Iterator {
 	return &Iterator{
 		Next:       make(chan record.Record, 10),
 		Done:       make(chan struct{}),
+		doneClosed: abool.NewBool(false),
 	}
 }
 
+// Finish is called be the storage to signal the end of the query results.
 func (it *Iterator) Finish(err error) {
 	close(it.Next)
+	if it.doneClosed.SetToIf(false, true) {
 		close(it.Done)
-	it.Err = err
+	}
+
+	it.errLock.Lock()
+	defer it.errLock.Unlock()
+	it.err = err
+}
+
+// Cancel is called by the iteration consumer to cancel the running query.
+func (it *Iterator) Cancel() {
+	if it.doneClosed.SetToIf(false, true) {
+		close(it.Done)
+	}
+}
+
+// Err returns the iterator error, if exists.
+func (it *Iterator) Err() error {
+	it.errLock.Lock()
+	defer it.errLock.Unlock()
+	return it.err
 }

database/maintenance.go

@@ -64,7 +64,7 @@ func MaintainRecordStates() error {
 				toExpire = append(toExpire, r)
 			}
 		}
-    if it.Err != nil {
+		if it.Err() != nil {
 			return err
 		}
 

database/record/base.go

@@ -21,6 +21,11 @@ func (b *Base) Key() string {
 	return fmt.Sprintf("%s:%s", b.dbName, b.dbKey)
 }
 
+// KeyIsSet returns true if the database key is set.
+func (b *Base) KeyIsSet() bool {
+	return len(b.dbName) > 0 && len(b.dbKey) > 0
+}
+
 // DatabaseName returns the name of the database.
 func (b *Base) DatabaseName() string {
 	return b.dbName
@@ -47,6 +52,11 @@ func (b *Base) Meta() *Meta {
 	return b.meta
 }
 
+// CreateMeta sets a default metadata object for this record.
+func (b *Base) CreateMeta() {
+	b.meta = &Meta{}
+}
+
 // SetMeta sets the metadata on the database record, it should only be called after loading the record. Use MoveTo to save the record with another key.
 func (b *Base) SetMeta(meta *Meta) {
 	b.meta = meta

database/record/meta.go

@@ -78,6 +78,11 @@ func (m *Meta) Delete() {
 	m.Deleted = time.Now().Unix()
 }
 
+// IsDeleted returns whether the record is deleted.
+func (m *Meta) IsDeleted() bool {
+	return m.Deleted > 0
+}
+
 // CheckValidity checks whether the database record is valid.
 func (m *Meta) CheckValidity() (valid bool) {
 	switch {

database/record/record.go

@@ -7,6 +7,7 @@ import (
 // Record provides an interface for uniformally handling database records.
 type Record interface {
 	Key() string // test:config
+	KeyIsSet() bool
 	DatabaseName() string // test
 	DatabaseKey() string  // config
 

database/storage/badger/badger.go

@@ -121,7 +121,11 @@ func (b *Badger) queryExecutor(queryIter *iterator.Iterator, q *query.Query, loc
 		for it.Seek(prefix); it.ValidForPrefix(prefix); it.Next() {
 			item := it.Item()
 
-			data, err := item.Value()
+			var data []byte
+			err := item.Value(func(val []byte) error {
+				data = val
+				return nil
+			})
 			if err != nil {
 				return err
 			}
@@ -148,10 +152,14 @@ func (b *Badger) queryExecutor(queryIter *iterator.Iterator, q *query.Query, loc
 					return err
 				}
 				select {
+				case <-queryIter.Done:
+					return nil
 				case queryIter.Next <- new:
 				default:
 					select {
 					case queryIter.Next <- new:
+					case <-queryIter.Done:
+						return nil
 					case <-time.After(1 * time.Minute):
 						return errors.New("query timeout")
 					}
@@ -162,11 +170,7 @@ func (b *Badger) queryExecutor(queryIter *iterator.Iterator, q *query.Query, loc
 		return nil
 	})
 
-	if err != nil {
+	queryIter.Finish(err)
-		queryIter.Err = err
-	}
-	close(queryIter.Next)
-	close(queryIter.Done)
 }
 
 // ReadOnly returns whether the database is read only.

modules/modules.go

@@ -24,7 +24,7 @@ var (
 type Module struct {
 	Name         string
 	Active       *abool.AtomicBool
-	inTransition bool
+	inTransition *abool.AtomicBool
 
 	prep  func() error
 	start func() error
@@ -42,6 +42,7 @@ func Register(name string, prep, start, stop func() error, dependencies ...strin
 	newModule := &Module{
 		Name:         name,
 		Active:       abool.NewBool(false),
+		inTransition: abool.NewBool(false),
 		prep:         prep,
 		start:        start,
 		stop:         stop,

modules/modules_test.go

@@ -5,8 +5,8 @@ package modules
 import (
 	"errors"
 	"fmt"
-	"testing"
 	"sync"
+	"testing"
 	"time"
 )
 
@@ -79,7 +79,7 @@ func TestOrdering(t *testing.T) {
 func resetModules() {
 	for _, module := range modules {
 		module.Active.UnSet()
-		module.inTransition = false
+		module.inTransition.UnSet()
 	}
 }
 

modules/start.go

@@ -72,7 +72,7 @@ moduleLoop:
 		switch {
 		case module.Active.IsSet():
 			active++
-		case module.inTransition:
+		case module.inTransition.IsSet():
 			modulesInProgress = true
 		default:
 			for _, depName := range module.dependencies {
@@ -116,7 +116,7 @@ func startModules() error {
 
 		for _, module := range readyToStart {
 			modulesStarting.Add(1)
-			module.inTransition = true
+			module.inTransition.Set()
 			nextModule := module // workaround go vet alert
 			go func() {
 				startErr := nextModule.start()
@@ -125,7 +125,7 @@ func startModules() error {
 				} else {
 					log.Infof("modules: started %s", nextModule.Name)
 					nextModule.Active.Set()
-					nextModule.inTransition = false
+					nextModule.inTransition.UnSet()
 					reports <- nil
 				}
 				modulesStarting.Done()

modules/stop.go

@@ -42,7 +42,7 @@ func checkStopStatus() (readyToStop []*Module, done bool) {
 
 	// make list out of map, minus modules in transition
 	for _, module := range activeModules {
-		if !module.inTransition {
+		if !module.inTransition.IsSet() {
 			readyToStop = append(readyToStop, module)
 		}
 	}
@@ -74,7 +74,7 @@ func Shutdown() error {
 		}
 
 		for _, module := range readyToStop {
-			module.inTransition = true
+			module.inTransition.Set()
 			nextModule := module // workaround go vet alert
 			go func() {
 				err := nextModule.stop()
@@ -84,7 +84,7 @@ func Shutdown() error {
 					reports <- nil
 				}
 				nextModule.Active.UnSet()
-				nextModule.inTransition = false
+				nextModule.inTransition.UnSet()
 			}()
 		}
 

utils/slices.go

@@ -1,22 +1,25 @@
 package utils
 
+// IndexOfString returns the index of given string and -1 if its not part of the slice.
+func IndexOfString(a []string, s string) int {
+	for i, entry := range a {
+		if entry == s {
+			return i
+		}
+	}
+	return -1
+}
+
 // StringInSlice returns whether the given string is in the string slice.
 func StringInSlice(a []string, s string) bool {
-	for _, entry := range a {
+	return IndexOfString(a, s) >= 0
-		if entry == s {
-			return true
-		}
-	}
-	return false
 }
 
 // RemoveFromStringSlice removes the given string from the slice and returns a new slice.
 func RemoveFromStringSlice(a []string, s string) []string {
-	for key, entry := range a {
+	i := IndexOfString(a, s)
-		if entry == s {
+	if i > 0 {
-			a = append(a[:key], a[key+1:]...)
+		a = append(a[:i], a[i+1:]...)
-			return a
-		}
 	}
 	return a
 }