package main

import (
	"context"
	"fmt"

	"cloud.google.com/go/pubsub"
)

// Receiver wraps a pubsub.Receiver.
type Receiver struct {
	Subscription *pubsub.Subscription
}

// NewReceiver creates a new Receiver.
func NewReceiver(ctx context.Context, project, subscription string) (*Receiver, error) {
	c, err := pubsub.NewClient(ctx, project)
	if err != nil {
		return nil, err
	}
	s := c.Subscription(subscription)
	ok, err := s.Exists(ctx)
	if err != nil {
		return nil, err
	}
	if !ok {
		return nil, fmt.Errorf("subscription %q does not exist", subscription)
	}

	return &Receiver{
		Subscription: s,
	}, nil
}

// Receive receives pubsub messages.
func (r *Receiver) Receive(ctx context.Context, f func(context.Context, *pubsub.Message) error) error {
	return r.Subscription.Receive(ctx, func(ctx context.Context, m *pubsub.Message) {
		r.Process(ctx, m, f)
	})
}

// Process wraps the processing of a single pubsub message through the provided callback.
// Process automatically Ack or Nack the message based on the value returned by the callback.
func (r *Receiver) Process(ctx context.Context, m *pubsub.Message, f func(context.Context, *pubsub.Message) error) {
	err := f(ctx, m)
	if err != nil {
		res := m.NackWithResult()
		status, err := res.Get(ctx)
		fmt.Println(status, err)
		return
	}

	res := m.AckWithResult()
	status, err := res.Get(ctx)
	fmt.Println(status, err)
}

func TestProcess(t *testing.T) {
	r := &Receiver{}
	msg := &pubsub.Message{
		Data: []byte("foo"),
	}

	ctx := context.Background()
	r.Process(ctx, msg, func(ctx context.Context, m *pubsub.Message) error {
		fmt.Println(string(msg.Data))
		return nil
	})
}