Staging

.drone.yml

@@ -3,7 +3,7 @@ env:
   - GOPATH=/var/cache/drone
 services:
   - redis
-  - apcera/gnatsd
+  - mikedewar/nsqd
   - dockerfile/rethinkdb
 script:
   - pip install fabric
@@ -18,9 +18,9 @@ script:
   - "if [ \"$DRONE_BRANCH\" = \"develop\" ]; then fab -H bart.lavaboom.io:36467 deploy; fi"
 notify:
   slack:
-   - webhook_url: $$SLACK_URL
-   - channel: $$SLACK_CHANNEL
-   - username: lavadrone
-   - on_started: true
-   - on_success: true
-   - on_failure: true
+    webhook_url: $$SLACK_URL
+    channel: $$SLACK_CHANNEL
+    username: lavadrone
+    on_started: true
+    on_success: true
+    on_failure: true

db/setup.go

@@ -14,13 +14,13 @@ var (
 var tableIndexes = map[string][]string{
 	"accounts":     []string{"name"},
 	"contacts":     []string{"owner"},
-	"emails":       []string{"owner", "label_ids"},
+	"emails":       []string{"owner", "label_ids", "date_created", "thread"},
+	"files":        []string{"owner"},
 	"keys":         []string{"owner", "key_id"},
 	"labels":       []string{"owner"},
 	"reservations": []string{"email", "name"},
-	"threads":      []string{"owner"},
+	"threads":      []string{"owner", "subject_hash"},
 	"tokens":       []string{"owner"},
-	"attachments":  []string{"owner"},
 }
 
 // List of names of databases

db/table_emails.go

@@ -65,7 +65,7 @@ func (e *EmailsTable) List(
 		filter["thread"] = thread
 	}
 
-	term := e.GetTable().Filter(filter)
+	term := e.GetTable().Filter(filter).Filter(gorethink.Not(gorethink.Row.Field("status").Eq(gorethink.Expr("queued"))))
 
 	// If sort array has contents, parse them and add to the term
 	if sort != nil && len(sort) > 0 {
@@ -128,3 +128,24 @@ func (e *EmailsTable) DeleteByThread(id string) error {
 		"thread": id,
 	})
 }
+
+func (e *EmailsTable) GetThreadManifest(thread string) (string, error) {
+	cursor, err := e.GetTable().
+		GetAllByIndex("thread", thread).
+		OrderBy("date_created").
+		Limit(1).
+		Pluck("manifest").
+		Field("manifest").
+		Run(e.GetSession())
+	if err != nil {
+		return "", err
+	}
+
+	var manifest string
+	err = cursor.One(&manifest)
+	if err != nil {
+		return "", err
+	}
+
+	return manifest, nil
+}

db/table_files.go

@@ -0,0 +1,95 @@
+package db
+
+import (
+	"github.com/lavab/api/models"
+
+	"github.com/dancannon/gorethink"
+)
+
+type FilesTable struct {
+	RethinkCRUD
+	Emails *EmailsTable
+}
+
+func (f *FilesTable) GetFile(id string) (*models.File, error) {
+	var result models.File
+
+	if err := f.FindFetchOne(id, &result); err != nil {
+		return nil, err
+	}
+
+	return &result, nil
+}
+
+func (f *FilesTable) GetOwnedBy(id string) ([]*models.File, error) {
+	var result []*models.File
+
+	err := f.WhereAndFetch(map[string]interface{}{
+		"owner": id,
+	}, &result)
+	if err != nil {
+		return nil, err
+	}
+
+	return result, nil
+}
+
+func (f *FilesTable) DeleteOwnedBy(id string) error {
+	return f.Delete(map[string]interface{}{
+		"owner": id,
+	})
+}
+
+func (f *FilesTable) GetEmailFiles(id string) ([]*models.File, error) {
+	email, err := f.Emails.GetEmail(id)
+	if err != nil {
+		return nil, err
+	}
+
+	query, err := f.Emails.GetTable().Filter(func(row gorethink.Term) gorethink.Term {
+		return gorethink.Expr(email.Files).Contains(row.Field("id"))
+	}).GetAll().Run(f.Emails.GetSession())
+	if err != nil {
+		return nil, err
+	}
+
+	var result []*models.File
+	err = query.All(&result)
+	if err != nil {
+		return nil, err
+	}
+
+	return result, nil
+}
+
+func (f *FilesTable) CountByEmail(id string) (int, error) {
+	query, err := f.GetTable().GetAllByIndex("owner", id).Count().Run(f.GetSession())
+	if err != nil {
+		return 0, err
+	}
+
+	var result int
+	err = query.One(&result)
+	if err != nil {
+		return 0, err
+	}
+
+	return result, nil
+}
+
+func (f *FilesTable) CountByThread(id ...interface{}) (int, error) {
+	query, err := f.GetTable().Filter(func(row gorethink.Term) gorethink.Term {
+		return gorethink.Table("emails").GetAllByIndex("thread", id...).Field("files").Contains(row.Field("id"))
+	}).Count().Run(f.GetSession())
+	if err != nil {
+		return 0, err
+	}
+
+	var result int
+	err = query.One(&result)
+	if err != nil {
+		return 0, err
+	}
+
+	return result, nil
+}

db/table_threads.go

@@ -60,18 +60,16 @@ func (t *ThreadsTable) List(
 				row.Field("labels").Contains(label),
 			)
 		})
-	}
-
-	if owner != "" && label == "" {
+	} else if owner != "" && label == "" {
 		term = t.GetTable().Filter(map[string]interface{}{
 			"owner": owner,
 		})
-	}
-
-	if owner == "" && label != "" {
+	} else if owner == "" && label != "" {
 		term = t.GetTable().Filter(func(row gorethink.Term) gorethink.Term {
 			return row.Field("labels").Contains(label)
 		})
+	} else {
+		term = t.GetTable()
 	}
 
 	// If sort array has contents, parse them and add to the term
@@ -103,6 +101,15 @@ func (t *ThreadsTable) List(
 		term = term.Slice(offset, offset+limit)
 	}
 
+	// Add manifests
+	term = term.InnerJoin(gorethink.Db(t.GetDBName()).Table("emails").Pluck("thread", "manifest"), func(thread gorethink.Term, email gorethink.Term) gorethink.Term {
+		return thread.Field("id").Eq(email.Field("thread"))
+	}).Without(map[string]interface{}{
+		"right": map[string]interface{}{
+			"thread": true,
+		},
+	}).Zip()
+
 	// Run the query
 	cursor, err := term.Run(t.GetSession())
 	if err != nil {

env/config.go

@@ -18,7 +18,8 @@ type Flags struct {
 	RethinkDBKey      string
 	RethinkDBDatabase string
 
-	NATSAddress string
+	LookupdAddress string
+	NSQdAddress    string
 
 	YubiCloudID  string
 	YubiCloudKey string

env/env.go

@@ -2,7 +2,7 @@ package env
 
 import (
 	"github.com/Sirupsen/logrus"
-	"github.com/apcera/nats"
+	"github.com/bitly/go-nsq"
 	"github.com/dancannon/gorethink"
 
 	"github.com/lavab/api/cache"
@@ -33,12 +33,12 @@ var (
 	Emails *db.EmailsTable
 	// Labels is the global instance of LabelsTable
 	Labels *db.LabelsTable
-	// Attachments is the global instance of AttachmentsTable
-	Attachments *db.AttachmentsTable
+	// Files is the global instance of FilesTable
+	Files *db.FilesTable
 	// Threads is the global instance of ThreadsTable
 	Threads *db.ThreadsTable
 	// Factors contains all currently registered factors
 	Factors map[string]factor.Factor
-	// NATS is the encoded connection to the NATS queue
-	NATS *nats.EncodedConn
+	// Producer is the nsq producer used to send messages to other components of the system
+	Producer *nsq.Producer
 )