remark42/backend/app/rest/api/migrator.go

package api

import (
	"bytes"
	"compress/gzip"
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	cache "github.com/go-pkgz/lcw/v2"
	log "github.com/go-pkgz/lgr"
	R "github.com/go-pkgz/rest"

	"github.com/umputun/remark42/backend/app/migrator"
	"github.com/umputun/remark42/backend/app/rest"
	"github.com/umputun/remark42/backend/app/store/engine"
)

// Migrator rest with import and export controllers
type Migrator struct {
	Cache             LoadingCache
	NativeImporter    migrator.Importer
	DisqusImporter    migrator.Importer
	WordPressImporter migrator.Importer
	CommentoImporter  migrator.Importer
	NativeExporter    migrator.Exporter
	URLMapperMaker    migrator.MapperMaker
	KeyStore          KeyStore

	busy map[string]bool
	lock sync.Mutex
}

// KeyStore defines sub-interface for consumers needed just a key
type KeyStore interface {
	Key(siteID string) (key string, err error)
}

// POST /import?secret=key&site=site-id&provider=disqus|remark|wordpress
// imports comments from post body.
func (m *Migrator) importCtrl(w http.ResponseWriter, r *http.Request) {
	siteID := r.URL.Query().Get("site")

	if m.isBusy(siteID) {
		rest.SendErrorJSON(w, r, http.StatusConflict, fmt.Errorf("already running"),
			"import rejected", rest.ErrActionRejected)
		return
	}

	tmpfile, err := m.saveTemp(r.Body)
	if err != nil {
		rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't save request to temp file", rest.ErrInternal)
		return
	}

	go m.runImport(siteID, r.URL.Query().Get("provider"), tmpfile) // import runs in background and sets busy flag for site

	_ = R.EncodeJSON(w, http.StatusAccepted, R.JSON{"status": "import request accepted"})
}

// POST /import/form?secret=key&site=site-id&provider=disqus|remark|wordpress
// imports comments from form body.
func (m *Migrator) importFormCtrl(w http.ResponseWriter, r *http.Request) {
	siteID := r.URL.Query().Get("site")

	if m.isBusy(siteID) {
		rest.SendErrorJSON(w, r, http.StatusConflict, fmt.Errorf("already running"),
			"import rejected", rest.ErrActionRejected)
		return
	}

	r.Body = http.MaxBytesReader(w, r.Body, 256*1024*1024) // hard cap on upload to prevent memory exhaustion
	reader, err := r.MultipartReader()
	if err != nil {
		rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't parse multipart form", rest.ErrDecode)
		return
	}

	tmpfile := ""
	for {
		part, err := reader.NextPart()
		if err == io.EOF {
			break
		}
		if err != nil {
			rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't parse multipart form", rest.ErrDecode)
			return
		}
		if part.FormName() != "file" {
			_ = part.Close()
			continue
		}

		tmpfile, err = m.saveTemp(part)
		if closeErr := part.Close(); err == nil && closeErr != nil {
			err = closeErr
		}
		if err != nil {
			rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't save request to temp file", rest.ErrInternal)
			return
		}
		break
	}
	if tmpfile == "" {
		rest.SendErrorJSON(w, r, http.StatusInternalServerError, fmt.Errorf("file field missing"),
			"can't get import file from the request", rest.ErrInternal)
		return
	}

	go m.runImport(siteID, r.URL.Query().Get("provider"), tmpfile) // import runs in background and sets busy flag for site

	_ = R.EncodeJSON(w, http.StatusAccepted, R.JSON{"status": "import request accepted"})
}

// GET /wait?site=site-id
// waits for migration operation (import or remap)
func (m *Migrator) waitCtrl(w http.ResponseWriter, r *http.Request) {
	siteID := r.URL.Query().Get("site")
	timeOut := time.Minute * 15
	if v := r.URL.Query().Get("timeout"); v != "" {
		if vv, e := time.ParseDuration(v); e == nil {
			timeOut = vv
		}
	}

	ctx, cancel := context.WithTimeout(context.Background(), timeOut)
	defer cancel()
	for m.isBusy(siteID) {

		select {
		case <-ctx.Done():
			_ = R.EncodeJSON(w, http.StatusGatewayTimeout, R.JSON{"status": "timeout expired", "site_id": siteID})
			return
		case <-time.After(100 * time.Millisecond):
		}
	}
	R.RenderJSON(w, R.JSON{"status": "completed", "site_id": siteID})
}

// GET /export?site=site-id&secret=12345&?mode=file|stream
// exports all comments for siteID as gz file
func (m *Migrator) exportCtrl(w http.ResponseWriter, r *http.Request) {
	siteID := r.URL.Query().Get("site")

	if r.URL.Query().Get("mode") == "file" {
		// buffer to memory to handle errors before committing to response
		var buf bytes.Buffer
		gzWriter := gzip.NewWriter(&buf)
		if _, err := m.NativeExporter.Export(gzWriter, siteID); err != nil {
			code, errCode := exportErrStatus(err)
			rest.SendErrorJSON(w, r, code, err, "export failed", errCode)
			return
		}
		if err := gzWriter.Close(); err != nil {
			rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "export failed", rest.ErrInternal)
			return
		}

		exportFile := fmt.Sprintf("%s-%s.json.gz", siteID, time.Now().Format("20060102"))
		w.Header().Set("Content-Type", "application/gzip")
		w.Header().Set("Content-Disposition", "attachment;filename="+exportFile)
		w.Header().Set("Content-Length", strconv.Itoa(buf.Len()))
		if _, err := io.Copy(w, &buf); err != nil {
			log.Printf("[WARN] failed to write export response: %v", err)
		}
		return
	}

	// stream mode - write directly to response
	if _, err := m.NativeExporter.Export(w, siteID); err != nil {
		code, errCode := exportErrStatus(err)
		rest.SendErrorJSON(w, r, code, err, "export failed", errCode)
	}
}

// exportErrStatus maps an export failure to an HTTP status and error code: an unknown
// site is a client error (400), anything else is treated as internal (500).
// The bolt store returns the engine.ErrSiteNotFound sentinel; the rpc store loses typed
// errors over jrpc, so the "not found" message is matched as a fallback (export only ever
// hits a site-level lookup, so a "not found" here can only mean the site).
func exportErrStatus(err error) (status, errCode int) {
	if errors.Is(err, engine.ErrSiteNotFound) || strings.Contains(err.Error(), "not found") {
		return http.StatusBadRequest, rest.ErrSiteNotFound
	}
	return http.StatusInternalServerError, rest.ErrInternal
}

// POST /remap?site=site-id
// remap urls in comments based on given rules (oldUrl newUrl)
func (m *Migrator) remapCtrl(w http.ResponseWriter, r *http.Request) {
	siteID := r.URL.Query().Get("site")

	// create new url-mapper from given rules in body
	mapper, err := m.URLMapperMaker(r.Body)
	if err != nil {
		rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "remap failed, bad given rules", rest.ErrDecode)
		return
	}
	defer r.Body.Close() //nolint gosec // we don't care about response body

	// start remap procedure with mapper
	go func() {
		m.setBusy(siteID, true)
		defer m.setBusy(siteID, false)

		// do export
		fh, e := os.CreateTemp("", "remark42_convert")
		if e != nil {
			log.Printf("[WARN] failed to make temp file %+v", e)
			return
		}
		defer func() {
			if e = os.Remove(fh.Name()); e != nil { //nolint:gosec // fh.Name() is from os.CreateTemp, server-controlled
				log.Printf("[WARN] failed to remove temp file %+v", e)
			}
		}()
		log.Printf("[DEBUG] start export for site=%s", siteID)
		if _, e = m.NativeExporter.Export(fh, siteID); e != nil {
			log.Printf("[WARN] export failed with %+v", e)
			return
		}

		if _, e = fh.Seek(0, 0); e != nil {
			log.Printf("[WARN] failed to seek file %+v", e)
			return
		}

		log.Printf("[DEBUG] start import for site=%s", siteID)
		mappedReader := migrator.WithMapper(fh, mapper)
		size, e := m.NativeImporter.Import(mappedReader, siteID)
		if e != nil {
			log.Printf("[WARN] import failed with %+v", e)
			return
		}

		m.Cache.Flush(cache.Flusher(siteID).Scopes(siteID))
		log.Printf("[DEBUG] convert request completed. site=%s, comments=%d", siteID, size)
	}()

	_ = R.EncodeJSON(w, http.StatusAccepted, R.JSON{"status": "convert request accepted"})
}

// runImport reads from tmpfile and import for given siteID and provider
func (m *Migrator) runImport(siteID, provider, tmpfile string) {
	m.setBusy(siteID, true)

	defer func() {
		m.setBusy(siteID, false)
		if err := os.Remove(tmpfile); err != nil {
			log.Printf("[WARN] failed to remove tmp file %s, %v", tmpfile, err)
		}
	}()

	var importer migrator.Importer
	switch provider {
	case "disqus":
		importer = m.DisqusImporter
	case "wordpress":
		importer = m.WordPressImporter
	case "commento":
		importer = m.CommentoImporter
	default:
		importer = m.NativeImporter
	}
	log.Printf("[DEBUG] import request for site=%s, provider=%s", siteID, provider)

	fh, err := os.Open(tmpfile) // nolint
	if err != nil {
		log.Printf("[WARN] import failed, %v", err)
		return
	}

	size, err := importer.Import(fh, siteID)
	if err != nil {
		log.Printf("[WARN] import failed, %v", err)
		return
	}
	m.Cache.Flush(cache.Flusher(siteID).Scopes(siteID))
	log.Printf("[DEBUG] import request completed. site=%s, provider=%s, comments=%d", siteID, provider, size)
}

// saveTemp reads from reader and saves to temp file
func (m *Migrator) saveTemp(r io.Reader) (string, error) {
	tmpfile, err := os.CreateTemp("", "remark42_import")
	if err != nil {
		return "", fmt.Errorf("can't make temp file: %w", err)
	}

	if _, err = io.Copy(tmpfile, r); err != nil {
		return "", fmt.Errorf("can't copy to temp file: %w", err)
	}

	if err = tmpfile.Close(); err != nil {
		return "", fmt.Errorf("can't close temp file: %w", err)
	}

	return tmpfile.Name(), nil
}

// isBusy checks busy flag from the map by siteID as key
func (m *Migrator) isBusy(siteID string) bool {
	m.lock.Lock()
	defer m.lock.Unlock()
	if m.busy == nil {
		m.busy = map[string]bool{}
	}
	return m.busy[siteID]
}

// setBusy sets/resets busy flag to the map by siteID as key
func (m *Migrator) setBusy(siteID string, status bool) {
	m.lock.Lock()
	defer m.lock.Unlock()
	if m.busy == nil {
		m.busy = map[string]bool{}
	}
	m.busy[siteID] = status
}