bit4sat/storage/upload_model.go

package storage

import (
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"

	"git.sp4ke.com/sp4ke/bit4sat/bus"
	"git.sp4ke.com/sp4ke/bit4sat/db"
	"git.sp4ke.com/sp4ke/bit4sat/ln"
	"github.com/jmoiron/sqlx"
	"github.com/lib/pq"
	"github.com/mediocregopher/radix/v3"
)

var DB = db.DB

const (
	//TODO: sync upload status from redis
	// TODO: status is currently handled in cache not here
	DBUploadSchema = `
		CREATE TABLE IF NOT EXISTS upload (
			id serial PRIMARY KEY,
			upload_id varchar(9) NOT NULL,
			sha256 varchar(64) NOT NULL,
			file_name varchar(255) NOT NULL,
			file_type varchar(255) DEFAULT '',
			file_size integer NOT NULL,
			file_ext varchar(255) DEFAULT '',
			fee integer NOT NULL DEFAULT 0,
			status integer DEFAULT 0 REFERENCES upload_status(type),
			UNIQUE (upload_id, sha256)
		);
	`

	DBUploadView = `
CREATE OR REPLACE VIEW upload_with_status
AS SELECT 
	upload.upload_id,
	upload.sha256,
	upload_status.status,
	upload.file_name,
	upload.file_type,
	upload.file_size,
	upload.file_ext,
	upload.fee
 FROM upload
JOIN upload_status ON
upload.status = upload_status.type
	`

	DBUploadStatusSchema = `CREATE TABLE IF NOT EXISTS upload_status
		(
		type INT NOT NULL PRIMARY KEY,
		status CHAR(255) NOT NULL UNIQUE
		)
	`

	QNewUpload = `INSERT INTO upload
	(upload_id, sha256, file_name, file_type, file_size, file_ext, status, fee)
	VALUES
	(:upload_id, :sha256, :file_name, :file_type, :file_size, :file_ext, :status, :fee)`

	QSetStatus = `UPDATE upload SET status = :status WHERE upload_id = :upload_id `

	QGetByHashID = `SELECT  upload_id,
							sha256,
							file_name,
							file_type,
							file_size,
							file_ext,
							fee,
							status
FROM upload WHERE sha256 = $1 AND upload_id = $2`
)

const (
	UpNew = iota
	UpWaitingPayment
	UpWaitingStorage
	UpStored
	UpReady
)

// new --> waiting payment (optional) --> waiting storage (client upload)
// --> Sotred --> Ready (sharing url sent to client and acked)
var UploadStatus = map[int]string{
	UpNew:            "new upload",
	UpWaitingPayment: "waiting payment",
	UpWaitingStorage: "waiting storage",
	UpStored:         "stored",
	UpReady:          "ready",
}

var (
	ErrDoesNotExist  = errors.New("does not exist")
	ErrAlreadyExists = errors.New("already exists")
)

type Upload struct {
	ID       string `db:"upload_id"`
	SHA256   string `db:"sha256"`
	FileName string `db:"file_name"`
	FileType string `db:"file_type"`
	FileSize int64  `db:"file_size"`
	FileExt  string `db:"file_ext"`
	Status   int    `db:"status"`
	Fee      int    `db:"fee"`
}

// TODO: sync from redis to db
//func SyncUploadStatusToDB(){

//}

func SetUploadStatus(id string, status int) error {
	key := fmt.Sprintf("upload_status_%s", id)
	return DB.Redis.Do(radix.FlatCmd(nil, "SET", key, status))
}

func SetUploadInvoice(uploadId string, invoice *ln.Invoice) error {
	uploadInvoiceKey := fmt.Sprintf("upload_%s_invoice", uploadId)

	invoiceJson, err := json.Marshal(invoice)
	if err != nil {
		return err
	}

	err = DB.Redis.Do(radix.FlatCmd(nil, "SET", uploadInvoiceKey, invoiceJson))
	if err != nil {
		return err
	}

	// Set inverse relation
	invoiceUploadKey := fmt.Sprintf("invoice_%s_upload", invoice.Id)
	return DB.Redis.Do(radix.FlatCmd(nil, "SET", invoiceUploadKey, uploadId))
}

func WatchUploadPayment(invoice *ln.Invoice) error {

	log.Println("watching invoice ", invoice)

	invoiceJson, err := json.Marshal(invoice)
	if err != nil {
		return err
	}

	return DB.Redis.Do(radix.FlatCmd(nil, "PUBLISH", bus.WatchInvoicesChannelName, invoiceJson))

	//TODO: Notify client on websocket

	return nil
}

// Returns true if id exists in DB
func IdExists(id string) (exists bool, err error) {
	key := fmt.Sprintf("upload_status_%s", id)

	err = DB.Redis.Do(radix.Cmd(&exists, "EXISTS", key))

	return
}

// Get a file by upload id and hash
func GetByHashID(sha256 string, id string) (*Upload, error) {
	var up Upload

	err := DB.Sql.Get(&up, QGetByHashID, sha256, id)

	if err == sql.ErrNoRows {
		return nil, ErrDoesNotExist
	}

	if err != nil {
		return nil, err
	}

	return &up, nil
}

func (u *Upload) TxSetState(tx *sqlx.Tx, status int) error {
	_, err := tx.NamedExec(QSetStatus, u)

	if err != nil {
		return err
	}

	return nil
}

func (u *Upload) TxWrite(tx *sqlx.Tx) error {

	_, err := tx.NamedExec(QNewUpload, u)

	if pqError, ok := err.(*pq.Error); ok {
		// unique constraint
		if pqError.Code == "23505" {
			return ErrAlreadyExists
		}
	}

	if err != nil {
		return err
	}

	return nil
}

func (u *Upload) Write() error {
	_, err := DB.Sql.NamedExec(QNewUpload, u)

	if pqError, ok := err.(*pq.Error); ok {
		// unique constraint
		if pqError.Code == "23505" {
			return ErrAlreadyExists
		}
	}

	if err != nil {
		return err
	}

	return nil
}

func init() {
	_, err := DB.Sql.Exec(DBUploadStatusSchema)
	if err != nil {
		log.Fatal(err)
	}

	_, err = DB.Sql.Exec(DBUploadSchema)
	if err != nil {
		log.Fatal(err)
	}

	_, err = DB.Sql.Exec(DBUploadView)
	if err != nil {
		log.Fatal(err)
	}

	// Populate status types
	query := `INSERT INTO upload_status (type, status)
	 VALUES($1,$2) ON CONFLICT (type) DO UPDATE
				SET status = $3`
	for k, v := range UploadStatus {
		_, err := DB.Sql.Exec(query, k, v, v)
		if err != nil {
			log.Panic(err)
		}
	}

}