ygggo/internal/smtpsender/sender.go

/*
 *  Original Copyright (c) 2021 Neil Alexander)
 *
 *  This Source Code Form is subject to the terms of the Mozilla Public
 *  License, v. 2.0. If a copy of the MPL was not distributed with this
 *  file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */

package smtpsender

import (
	"crypto/ed25519"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"log"
	"net/mail"
	"sync"
	"time"

	"github.com/neilalexander/yggmail/internal/config"
	"github.com/neilalexander/yggmail/internal/notify"
	"github.com/neilalexander/yggmail/internal/remote"
	pb "github.com/neilalexander/yggmail/internal/remote/types"
	"github.com/neilalexander/yggmail/internal/storage"
	"github.com/neilalexander/yggmail/internal/storage/types"
	"github.com/neilalexander/yggmail/internal/transport"
	"github.com/neilalexander/yggmail/internal/utils"
	"go.uber.org/atomic"
)

type Queues struct {
	Config    *config.Config
	Log       *log.Logger
	Transport transport.Transport
	Storage   storage.Storage
	RPCClient *remote.Client
	Notify    *notify.Notify
	queues    sync.Map
}

func NewQueues(config *config.Config, log *log.Logger, transport transport.Transport, storage storage.Storage, rpcClient *remote.Client, notify *notify.Notify) *Queues {
	qs := &Queues{
		Config:    config,
		Log:       log,
		Transport: transport,
		Storage:   storage,
		RPCClient: rpcClient,
		Notify:    notify,
	}
	time.AfterFunc(time.Second*5, qs.manager)
	return qs
}

func (qs *Queues) manager() {
	destinations, err := qs.Storage.QueueListDestinations()
	if err != nil {
		return
	}
	for _, destination := range destinations {
		_ = qs.QueueForServer(destination)
	}
	time.AfterFunc(time.Minute, qs.manager)
}

func (qs *Queues) QueueFor(storage storage.Storage, senderPK string, from string, rcpts []string, m io.Reader, size int) error {
	pid, err := storage.MailCreate(senderPK, "Outbox", m, size)
	if err != nil {
		return fmt.Errorf("storage.MailCreate(Outbox): %w", err)
	}

	for _, rcpt := range rcpts {
		addr, err := mail.ParseAddress(rcpt)
		if err != nil {
			continue
		}

		dPk, _, err := utils.ParseAddress(addr.Address)
		if err != nil {
			continue
		}

		hostHex := hex.EncodeToString(dPk)
		err = storage.QueueInsertDestinationForID(senderPK, hostHex, pid, from, rcpt)
		if err != nil {
			qs.Log.Printf("Failed to insert queue for %s: %v", hostHex, err)
			continue
		}
		if storage == qs.Storage {
			_ = qs.QueueForServer(hostHex)
		}
	}

	return nil
}

func (qs *Queues) QueueForServer(server string) error {
	v, _ := qs.queues.LoadOrStore(server, &Queue{
		queues:      qs,
		destination: server,
	})
	q, ok := v.(*Queue)
	if !ok {
		return fmt.Errorf("type assertion error")
	}
	if q.running.CompareAndSwap(false, true) {
		go q.run()
	}
	return nil
}

type Queue struct {
	queues *Queues
	// hex16
	destination string
	running     atomic.Bool
}

func (q *Queue) run() {
	defer q.running.Store(false)
	for {
		refs, err := q.queues.Storage.QueueMailIDsForDestination(q.destination)
		if err != nil || len(refs) == 0 {
			return
		}

		for _, ref := range refs {
			_, mail, err := q.queues.Storage.MailSelect(ref.AccountPubKey, "Outbox", ref.ID)
			if err != nil {
				q.handleError(ref, err)
				continue
			}

			q.queues.Log.Printf("Sending mail from %s (Acc: %s) to %s", ref.From, ref.AccountPubKey, q.destination)

			deliver := q.deliverRemote
			if q.destination == hex.EncodeToString(q.queues.Config.PublicKey) {
				deliver = q.deliverLocal
			}

			if err := deliver(ref, mail.Body, mail.Size); err != nil {
				q.queues.Log.Printf("Delivery to %s failed: %v. Will retry later.", q.destination, err)
				return
			}

			q.finalize(ref)
		}
	}
}

func (q *Queue) deliverLocal(ref types.QueuedMail, content io.ReadCloser, size uint32) error {
	q.queues.Log.Printf("Local delivery for %s", ref.Rcpt)

	_, toAcc, err := utils.ParseAddress(ref.Rcpt)
	toAccPk := utils.EncodeString(toAcc)
	if err != nil {
		return fmt.Errorf("local delivery: invalid rcpt address: %w", err)
	}
	if acc, err := q.queues.Storage.AccountGet(toAccPk); acc == nil && err != nil {
		return fmt.Errorf("local delivery: account not found")
	}

	_ = q.queues.Storage.MailboxCreate(toAccPk, "INBOX")
	id, err := q.queues.Storage.MailCreate(toAccPk, "INBOX", content, int(size))
	if err != nil {
		return fmt.Errorf("local delivery failed: %w", err)
	}
	if count, err := q.queues.Storage.MailCount(toAccPk, "INBOX"); err == nil {
		q.queues.Notify.NotifyNew(toAccPk, notify.NewEmailEvent(toAccPk, "INBOX", int64(id), int64(count)))
	}

	return nil
}

// Deliver incapsulate SMTP-Transaction
func (q *Queue) deliverRemote(ref types.QueuedMail, content io.ReadCloser, size uint32) error {
	dPkb, err := hex.DecodeString(q.destination)
	dPk := ed25519.PublicKey(dPkb)

	resp, _, err := q.queues.RPCClient.Call(dPk, &pb.Request{
		Call: &pb.Request_MailPush{
			MailPush: &pb.MailPushRequest{
				From: ref.From,
				To:   ref.Rcpt,
			},
		},
	}, content, size)

	if err != nil {
		return fmt.Errorf("rpc call failed: %w", err)
	}

	if !resp.Success {
		return fmt.Errorf("remote server returned error: %s", resp.Error)
	}

	q.queues.Log.Printf("Sender: RPC delivery successful for ID %d to node %s", ref.ID, q.destination)
	return nil
}

func (q *Queue) finalize(ref types.QueuedMail) {
	// Delete id from Queue
	_ = q.queues.Storage.QueueDeleteDestinationForID(ref.AccountPubKey, q.destination, ref.ID)

	remaining, err := q.queues.Storage.QueueSelectIsMessagePendingSend(ref.AccountPubKey, "Outbox", ref.ID)
	if err == nil && !remaining {
		q.queues.Log.Printf("Moving mail %d to Sent for %s", ref.ID, ref.AccountPubKey)
		_, _ = q.queues.Storage.MailMove(ref.AccountPubKey, "Outbox", ref.ID, "Sent")
	}

	_ = q.queues.Storage.MailExpunge(ref.AccountPubKey, "Outbox")
}

func (q *Queue) handleError(ref types.QueuedMail, err error) {
	if errors.Is(err, sql.ErrNoRows) {
		q.queues.Log.Printf("Cleaning up orphaned queue ref %d for %s", ref.ID, ref.AccountPubKey)
		_ = q.queues.Storage.QueueDeleteDestinationForID(ref.AccountPubKey, q.destination, ref.ID)
	} else {
		q.queues.Log.Printf("Storage error for %s: %v", ref.AccountPubKey, err)
	}
}