Fetch Activity events

management/server/activity/event.go

@@ -0,0 +1,85 @@
+package activity
+
+import "time"
+
+const (
+	// DeviceEvent describes an event that happened of a device (e.g, connected/disconnected)
+	DeviceEvent Type = "device"
+	// ManagementEvent describes an event that happened on a Management service (e.g., user added)
+	ManagementEvent Type = "management"
+)
+
+const (
+	AddPeerByUserOperation Operation = iota
+	AddPeerWithKeyOperation
+	UserJoinedOperation
+)
+
+const (
+	AddPeerByUserOperationMessage  string = "Add new peer"
+	AddPeerWithKeyOperationMessage string = AddPeerByUserOperationMessage
+	UserJoinedOperationMessage     string = "New user joined"
+)
+
+// MessageForOperation returns a string message for an Operation
+func MessageForOperation(op Operation) string {
+	switch op {
+	case AddPeerByUserOperation:
+		return AddPeerByUserOperationMessage
+	case AddPeerWithKeyOperation:
+		return AddPeerWithKeyOperationMessage
+	case UserJoinedOperation:
+		return UserJoinedOperationMessage
+	default:
+		return "UNKNOWN_OPERATION"
+	}
+}
+
+// Type of the Event
+type Type string
+
+// Operation is an action that triggered an Event
+type Operation int
+
+// Store provides an interface to store or stream events.
+type Store interface {
+	// Save an event in the store
+	Save(event *Event) (*Event, error)
+	// Get returns "limit" number of events from the "offset" index ordered descending or ascending by a timestamp
+	Get(accountID string, offset, limit int, descending bool) ([]*Event, error)
+	// Close the sink flushing events if necessary
+	Close() error
+}
+
+// Event represents a network/system activity event.
+type Event struct {
+	// Timestamp of the event
+	Timestamp time.Time
+	// Operation that was performed during the event
+	Operation string
+	// OperationCode that was performed during the event
+	OperationCode Operation
+	// ID of the event (can be empty, meaning that it wasn't yet generated)
+	ID uint64
+	// Type of the event
+	Type Type
+	// ModifierID is the ID of an object that modifies a Target
+	ModifierID string
+	// TargetID is the ID of an object that a Modifier modifies
+	TargetID string
+	// AccountID where event happened
+	AccountID string
+}
+
+// Copy the event
+func (e *Event) Copy() *Event {
+	return &Event{
+		Timestamp:  e.Timestamp,
+		Operation:  e.Operation,
+		ID:         e.ID,
+		Type:       e.Type,
+		ModifierID: e.ModifierID,
+		TargetID:   e.TargetID,
+		AccountID:  e.AccountID,
+	}
+}

management/server/activity/sqlite.go

@@ -0,0 +1,123 @@
+package activity
+
+import (
+	"database/sql"
+	"fmt"
+	_ "github.com/mattn/go-sqlite3"
+	"path/filepath"
+	"time"
+)
+
+const (
+	SQLiteEventSinkDB = "events.db"
+	createTableQuery  = "CREATE TABLE IF NOT EXISTS events " +
+		"(id INTEGER PRIMARY KEY AUTOINCREMENT, account TEXT NOT NULL, " +
+		"operation INTEGER, " +
+		"type TEXT, " +
+		"timestamp DATETIME, " +
+		"modifier TEXT," +
+		" target TEXT);"
+)
+
+// SQLiteStore is the implementation of the activity.Store interface backed by SQLite
+type SQLiteStore struct {
+	db *sql.DB
+}
+
+// NewSQLiteStore creates a new SQLiteStore with an event table if not exists.
+func NewSQLiteStore(dataDir string) (*SQLiteStore, error) {
+	dbFile := filepath.Join(dataDir, SQLiteEventSinkDB)
+	db, err := sql.Open("sqlite3", dbFile)
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = db.Exec(createTableQuery)
+	if err != nil {
+		return nil, err
+	}
+
+	return &SQLiteStore{db: db}, nil
+}
+
+func processResult(result *sql.Rows) ([]*Event, error) {
+	events := make([]*Event, 0)
+	for result.Next() {
+		var id int64
+		var operation Operation
+		var timestamp time.Time
+		var modifier string
+		var target string
+		var account string
+		var typ Type
+		err := result.Scan(&id, &operation, &timestamp, &modifier, &target, &account, &typ)
+		if err != nil {
+			return nil, err
+		}
+
+		events = append(events, &Event{
+			Timestamp:     timestamp,
+			OperationCode: operation,
+			Operation:     MessageForOperation(operation),
+			ID:            uint64(id),
+			Type:          typ,
+			ModifierID:    modifier,
+			TargetID:      target,
+			AccountID:     account,
+		})
+	}
+
+	return events, nil
+}
+
+// Get returns "limit" number of events from index ordered descending or ascending by a timestamp
+func (store *SQLiteStore) Get(accountID string, offset, limit int, descending bool) ([]*Event, error) {
+	order := "DESC"
+	if !descending {
+		order = "ASC"
+	}
+	stmt, err := store.db.Prepare(fmt.Sprintf("SELECT id, operation, timestamp, modifier, target, account, type"+
+		" FROM events WHERE account = ? ORDER BY timestamp %s LIMIT ? OFFSET ?;", order))
+	if err != nil {
+		return nil, err
+	}
+
+	result, err := stmt.Query(accountID, limit, offset)
+	if err != nil {
+		return nil, err
+	}
+
+	defer result.Close() //nolint
+	return processResult(result)
+}
+
+// Save an event in the SQLite events table
+func (store *SQLiteStore) Save(event *Event) (*Event, error) {
+
+	stmt, err := store.db.Prepare("INSERT INTO events(operation, timestamp, modifier, target, account, type) VALUES(?, ?, ?, ?, ?, ?)")
+	if err != nil {
+		return nil, err
+	}
+
+	result, err := stmt.Exec(event.OperationCode, event.Timestamp, event.ModifierID, event.TargetID, event.AccountID, event.Type)
+	if err != nil {
+		return nil, err
+	}
+
+	id, err := result.LastInsertId()
+	if err != nil {
+		return nil, err
+	}
+
+	eventCopy := event.Copy()
+	eventCopy.ID = uint64(id)
+	return eventCopy, nil
+}
+
+// Close the SQLiteStore
+func (store *SQLiteStore) Close() error {
+	if store.db != nil {
+		return store.db.Close()
+	}
+	return nil
+}

management/server/activity/sqlite_test.go

@@ -0,0 +1,52 @@
+package activity
+
+import (
+	"fmt"
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func TestNewSQLiteStore(t *testing.T) {
+	dataDir := t.TempDir()
+	store, err := NewSQLiteStore(dataDir)
+	if err != nil {
+		t.Fatal(err)
+		return
+	}
+
+	accountID := "account_1"
+
+	for i := 0; i < 10; i++ {
+		_, err = store.Save(&Event{
+			Timestamp:     time.Now(),
+			OperationCode: AddPeerByUserOperation,
+			Type:          ManagementEvent,
+			ModifierID:    "user_" + fmt.Sprint(i),
+			TargetID:      "peer_" + fmt.Sprint(i),
+			AccountID:     accountID,
+		})
+		if err != nil {
+			t.Fatal(err)
+			return
+		}
+	}
+
+	result, err := store.Get(accountID, 0, 10, false)
+	if err != nil {
+		t.Fatal(err)
+		return
+	}
+
+	assert.Len(t, result, 10)
+	assert.True(t, result[0].Timestamp.Before(result[len(result)-1].Timestamp))
+
+	result, err = store.Get(accountID, 0, 5, true)
+	if err != nil {
+		t.Fatal(err)
+		return
+	}
+
+	assert.Len(t, result, 5)
+	assert.True(t, result[0].Timestamp.After(result[len(result)-1].Timestamp))
+}