main.go

package main

import (
	"bytes"
	"encoding/base64"
	"encoding/json"
	"io"
	"log"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/go-stomp/stomp"
)

const TEST_RESOURCE = "http://localhost:8080/fcrepo/rest/test"

const TEST_TRIPLES = "./test-data/simple-container.nt"

// data structure for STOMP messages
type stompMessage struct {
	MsgId string   `json:"id"`
	Type  []string `json:"type"`
}

func createConn() *stomp.Conn {
	conn, err := stomp.Dial("tcp", "localhost:61613", stomp.ConnOpt.AcceptVersion(stomp.V12),
		stomp.ConnOpt.HeartBeat(10*time.Second, 5*time.Second))
	if err != nil {
		log.Fatalf("unable to connect to Fedora queue: %w", err)
	}
	log.Println("Connected to Fedora queue.")
	return conn
}

func loadTestTriples(path string) []byte {
	data, err := os.ReadFile(path)
	if err != nil {
		log.Fatalf("Test file not found at %s", path)
	}
	return data
}

func createSub(conn *stomp.Conn, done chan struct{}) chan *stomp.Message {
	messages := make(chan *stomp.Message)
	sub, err := conn.Subscribe("/queue/fedora", stomp.AckClient)
	if err != nil {
		log.Fatalf("Unable to subscibe to queue: %w", err)
	}
	log.Println("Subscribed to queue.")
	go func() {
		defer sub.Unsubscribe()
		defer close(messages)
		for {
			select {
			case msg := <-sub.C:
				if (msg != nil) && (msg.Body != nil) {
					messages <- msg
					if msg.ShouldAck() {
						err := conn.Ack(msg)
						if err != nil {
							log.Printf("Ack error, %+v\n", msg.Err.Error())
						}
					}
				}
			case <-done:
				return
			}
		}
	}()
	return messages
}

func makeRequest(client *http.Client, url, _type string, data []byte) *http.Response {
	auth := "fedoraAdmin:fedoraAdmin"
	auth = base64.StdEncoding.EncodeToString([]byte(auth))
	headers := map[string]string{"Authorization": "Basic " + auth, "Overwrite-Tombstone": "true"}
	if _type == "PUT" {
		headers["Content-Type"] = "text/turtle"
	}
	var req *http.Request
	if data != nil {
		req, _ = http.NewRequest(_type, url, bytes.NewBuffer(data))
	} else {
		req, _ = http.NewRequest(_type, url, nil)
	}
	for k, v := range headers {
		req.Header.Add(k, v)
	}
	resp, err := client.Do(req)
	if err != nil {
		log.Fatalf("Could not do %s request on Fedora API, %w", _type, err)
	}
	return resp
}

func deleteAndPurge(client *http.Client, url string) {
	log.Println("Deleting resource %s", url)
	resp := makeRequest(client, url, "DELETE", nil)
	if resp.StatusCode > 299 {
		body, _ := io.ReadAll(resp.Body)
		resp.Body.Close()
		log.Fatalf("Unable to delete test resource %s: %s", url, string(body))
	}
	// purge resource
	resp = makeRequest(client, url, "GET", nil)
	if resp.StatusCode == 410 {
		log.Println("Purging tombstone.")
		links := resp.Header["Link"]
		var tombstoneURL string
		for _, l := range links {
			if strings.Contains(l, "hasTombstone") {
				linkSlice := strings.Split(l, ";")
				tombstoneURL = strings.Trim(linkSlice[0], "<>")
				break
			}
		}
		resp = makeRequest(client, tombstoneURL, "DELETE", nil)
		if resp.StatusCode != 204 {
			body, _ := io.ReadAll(resp.Body)
			resp.Body.Close()
			log.Fatalf("Unable to delete tombstone %s: %s", url, string(body))
		}
	} else {
		log.Fatal(resp.Status)
	}
}

func createTestResource(client *http.Client) {
	// GET the resource: if it exists, we want to delete it first.
	resp := makeRequest(client, TEST_RESOURCE, "GET", nil)
	if resp.StatusCode < 300 { // exists -- delete first
		deleteAndPurge(client, TEST_RESOURCE)
	}
	testData := loadTestTriples(TEST_TRIPLES)
	resp = makeRequest(client, TEST_RESOURCE, "PUT", testData)
	body, _ := io.ReadAll(resp.Body)
	resp.Body.Close()
	if resp.StatusCode < 300 {
		// create successful
		log.Println("Created a resource!")
	} else {

		log.Fatalf("Unable to create the test resource %s: %s", TEST_RESOURCE, string(body))
	}
}

func main() {
	client := http.Client{Timeout: time.Duration(5) * time.Second}
	conn := createConn()
	done := make(chan struct{})
	messages := createSub(conn, done)
	// loop until we successfully create it or fail entirely
	createTestResource(&client)
	parts := strings.Split(TEST_RESOURCE, "/")
	resourceID := parts[len(parts)-1]
	// expect one message to show the resource has been created
	// should be the last message
loop:
	for {
		select {
		case msg := <-messages:
			if msg.Header.Get("org.fcrepo.jms.identifier") == "/"+resourceID {
				var parsedMessage stompMessage
				json.Unmarshal(msg.Body, &parsedMessage)
				if parsedMessage.Type[0] == "Create" {
					log.Println("Message received: object created.")
					break loop
				}
			}
		case <-time.After(10 * time.Second):
			log.Fatalf("Test failed. Message in queue not found for created object.")
		}
	}
	close(done)
	deleteAndPurge(&client, TEST_RESOURCE)
}