Merge pull request #13 from grvcoelho/deliveries

database: add deliveries

README.md

@@ -28,6 +28,5 @@
 | Attribute | Type | Description |
 | --------- | ---- | ----------- |
 | `status` | *String* | The status of the delivery. One of: `success`, `failed`, `processing` |
-| `latency` | *JSON* | The time between the delivery and the response from the client |
-| `response_status_code` | *String* | The HTTP status code the client respond with |
-| `response_headers` | *JSON* | The HTTP headers the client respond with |
+| `latency` | *Integer* | The time between the delivery and the response from the client |
+| `status_code` | *String* | The HTTP status code the client respond with |

migrations/1509740079_delivery_schema.down.sql

@@ -0,0 +1,2 @@
+DROP TYPE "deliveries_status";
+DROP TABLE IF EXISTS deliveries CASCADE;

migrations/1509740079_delivery_schema.up.sql

@@ -0,0 +1,15 @@
+CREATE TYPE "deliveries_status" AS ENUM (
+  'processing',
+  'failed',
+  'success'
+);
+
+CREATE TABLE IF NOT EXISTS deliveries (
+  id varchar(255) PRIMARY KEY,
+  message_id varchar(255) references messages(id) NOT NULL,
+  status "deliveries_status" DEFAULT 'processing'::"deliveries_status" NOT NULL,
+  latency integer,
+  status_code varchar(255),
+  created_at date NOT NULL,
+  updated_at date NOT NULL
+);

pkg/database/db_test.go

@@ -73,4 +73,32 @@ func TestDatabase(t *testing.T) {
 		assert.True(t, ok)
 		assert.Equal(t, m1.ID, m2.ID)
 	})
+
+	t.Run("DeliveryStore", func(t *testing.T) {
+		db, _ := NewDatabase(&cfg.Database{
+			Address: "postgres://postgres:postgres@database/webhulk?sslmode=disable",
+		})
+
+		webhookStore, _ := NewWebhookStore(db)
+		messageStore, _ := NewMessageStore(db)
+		deliveryStore, err := NewDeliveryStore(db)
+		assert.NoError(t, err)
+
+		w1, _ := NewWebhook("events", "https://receiver.com", true)
+		w1, _ = webhookStore.Store(w1)
+
+		m1, _ := NewMessage(w1.ID, []byte(`{"event":"status_changed"}`))
+		m1, _ = messageStore.Store(m1)
+
+		d1, err := NewDelivery(m1.ID)
+		assert.NoError(t, err)
+		assert.Equal(t, d1.Status, "processing")
+
+		_, err = deliveryStore.Store(d1)
+		assert.NoError(t, err, "Should store a delivery")
+
+		d2, ok := deliveryStore.Retrieve(d1.ID)
+		assert.True(t, ok)
+		assert.Equal(t, d1.ID, d2.ID)
+	})
 }

pkg/database/delivery.go

@@ -0,0 +1,17 @@
+package database
+
+type Delivery struct {
+	*Model
+	MessageID  string `db:"message_id"`
+	Status     string `db:"status"`
+	Latency    int    `db:"latency"`
+	StatusCode string `db:"status_code"`
+}
+
+func NewDelivery(messageID string) (*Delivery, error) {
+	return &Delivery{
+		Model:     NewModel("del"),
+		MessageID: messageID,
+		Status:    "processing",
+	}, nil
+}

pkg/database/delivery_store.go

@@ -0,0 +1,62 @@
+package database
+
+import (
+	"sync"
+
+	"github.com/jmoiron/sqlx"
+)
+
+type DeliveryStore struct {
+	db    *sqlx.DB
+	mutex sync.RWMutex
+}
+
+func NewDeliveryStore(db *sqlx.DB) (*DeliveryStore, error) {
+	return &DeliveryStore{
+		db:    db,
+		mutex: sync.RWMutex{},
+	}, nil
+}
+
+func (s *DeliveryStore) Store(d *Delivery) (*Delivery, error) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	query := `
+		INSERT INTO deliveries
+			(id, status, latency, status_code, message_id, created_at, updated_at)
+		VALUES
+			(:id, :status, :latency, :status_code, :message_id, :created_at, :updated_at)
+	`
+
+	_, err := s.db.NamedExec(query, d)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return d, nil
+}
+
+func (s *DeliveryStore) Retrieve(id string) (*Delivery, bool) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	d := &Delivery{}
+
+	query := `
+		SELECT
+			id, status, latency, status_code, message_id, created_at, updated_at
+		FROM deliveries
+		WHERE
+			id = $1
+	`
+
+	err := s.db.Get(d, query, id)
+
+	if err != nil {
+		return nil, false
+	}
+
+	return d, true
+}