next.orly.dev/pkg/database/import_utils.go

//go:build !(js && wasm)

// Package database provides shared import utilities for events
package database

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"os"
	"runtime/debug"
	"strings"
	"time"

	"git.mleku.dev/mleku/nostr/encoders/event"
	"lol.mleku.dev/chk"
	"lol.mleku.dev/log"
)

const maxLen = 500000000

// ImportEventsFromReader imports events from an io.Reader containing JSONL data
func (d *D) ImportEventsFromReader(ctx context.Context, rr io.Reader) error {
	startTime := time.Now()
	log.I.F("================================================")
	log.I.F("[HTTP API IMPORT] Starting import operation")
	log.I.F("[HTTP API IMPORT] State: Initializing - preparing to receive file data")
	log.I.F("================================================")

	// store to disk so we can return fast
	tmpPath := os.TempDir() + string(os.PathSeparator) + "orly"
	os.MkdirAll(tmpPath, 0700)
	tmp, err := os.CreateTemp(tmpPath, "")
	if chk.E(err) {
		log.E.F("[HTTP API IMPORT] ERROR: Failed to create temp file: %v", err)
		log.E.F("[HTTP API IMPORT] Action: Check disk space and temp directory permissions")
		return err
	}
	defer os.Remove(tmp.Name()) // Clean up temp file when done

	log.I.F("[HTTP API IMPORT] State: Buffering - receiving file data from client")
	log.I.F("[HTTP API IMPORT] Temp file: %s", tmp.Name())
	bufferStart := time.Now()
	bytesBuffered, err := io.Copy(tmp, rr)
	if chk.E(err) {
		log.E.F("[HTTP API IMPORT] ERROR: Failed to buffer file data: %v", err)
		log.E.F("[HTTP API IMPORT] Action: Check network connection and file integrity")
		return err
	}
	bufferElapsed := time.Since(bufferStart)
	fileSizeMB := float64(bytesBuffered) / 1024 / 1024
	bufferSpeedMBps := fileSizeMB / bufferElapsed.Seconds()
	log.I.F("[HTTP API IMPORT] State: Buffering complete")
	log.I.F("[HTTP API IMPORT] File size: %.2f MB | Buffer time: %v | Speed: %.2f MB/sec",
		fileSizeMB, bufferElapsed.Round(time.Millisecond), bufferSpeedMBps)

	if _, err = tmp.Seek(0, 0); chk.E(err) {
		log.E.F("[HTTP API IMPORT] ERROR: Failed to rewind temp file: %v", err)
		return err
	}

	// Estimate number of events (rough estimate: ~500 bytes per event average)
	estimatedEvents := int(float64(bytesBuffered) / 500)
	log.I.F("[HTTP API IMPORT] State: Processing - reading and importing events")
	log.I.F("[HTTP API IMPORT] Estimated events: ~%d (based on file size)", estimatedEvents)
	log.I.F("[HTTP API IMPORT] Progress updates will appear every 5 seconds")
	log.I.F("================================================")
	processErr := d.processJSONLEvents(ctx, tmp, bytesBuffered)

	totalElapsed := time.Since(startTime)
	log.I.F("================================================")
	log.I.F("[HTTP API IMPORT] State: Completed")
	log.I.F("[HTTP API IMPORT] Total time: %v", totalElapsed.Round(time.Millisecond))
	if processErr != nil {
		log.E.F("[HTTP API IMPORT] Result: FAILED")
		log.E.F("[HTTP API IMPORT] Error: %v", processErr)
		log.E.F("[HTTP API IMPORT] Action: Check error details above, verify file format is valid JSONL")
		log.I.F("================================================")
	} else {
		log.I.F("[HTTP API IMPORT] Result: SUCCESS")
		log.I.F("[HTTP API IMPORT] All events have been imported and are now available on the relay")
		log.I.F("================================================")
	}

	return processErr
}

// ImportEventsFromStrings imports events from a slice of JSON strings with policy filtering
func (d *D) ImportEventsFromStrings(ctx context.Context, eventJSONs []string, policyManager interface {
	CheckPolicy(action string, ev *event.E, pubkey []byte, remote string) (bool, error)
}) error {
	// Create a reader from the string slice
	reader := strings.NewReader(strings.Join(eventJSONs, "\n"))
	// Estimate total bytes (rough estimate)
	totalBytes := int64(len(strings.Join(eventJSONs, "\n")))
	return d.processJSONLEventsWithPolicy(ctx, reader, policyManager, totalBytes)
}

// processJSONLEvents processes JSONL events from a reader
func (d *D) processJSONLEvents(ctx context.Context, rr io.Reader, totalBytes int64) error {
	return d.processJSONLEventsWithPolicy(ctx, rr, nil, totalBytes)
}

// processJSONLEventsWithPolicy processes JSONL events from a reader with optional policy filtering
func (d *D) processJSONLEventsWithPolicy(ctx context.Context, rr io.Reader, policyManager interface {
	CheckPolicy(action string, ev *event.E, pubkey []byte, remote string) (bool, error)
}, totalBytes int64) error {
	// Create a scanner to read the buffer line by line
	scan := bufio.NewScanner(rr)
	scanBuf := make([]byte, maxLen)
	scan.Buffer(scanBuf, maxLen)

	// Performance tracking
	startTime := time.Now()
	lastLogTime := startTime
	const logInterval = 5 * time.Second

	var count, bytesRead, skipped, policyRejected, unmarshalErrors, saveErrors int
	var kind3Count, largeEventCount int // Track noisy event types

	// Track bytes read for progress percentage
	bytesReadTotal := int64(0)

	for scan.Scan() {
		select {
		case <-ctx.Done():
			log.I.F("[HTTP API IMPORT] State: Cancelled")
			log.I.F("[HTTP API IMPORT] Processed %d events before cancellation", count)
			return ctx.Err()
		default:
		}

		b := scan.Bytes()
		lineLen := len(b) + 1 // +1 for newline
		bytesReadTotal += int64(lineLen)
		bytesRead += lineLen

		if len(b) < 1 {
			skipped++
			continue
		}

		ev := event.New()
		if _, err := ev.Unmarshal(b); err != nil {
			// return the pooled buffer on error
			ev.Free()
			unmarshalErrors++
			// Only log unmarshal errors if they're frequent (not spam)
			if unmarshalErrors <= 10 || unmarshalErrors%100 == 0 {
				log.W.F("[HTTP API IMPORT] Failed to unmarshal event #%d: %v", unmarshalErrors, err)
			}
			continue
		}

		// Track event kinds for summary (but don't log each one)
		if ev.Kind == 3 {
			kind3Count++
		}
		if len(b) > 100000 { // Events larger than 100KB
			largeEventCount++
		}

		// Apply policy checking if policy manager is provided
		if policyManager != nil {
			// For sync imports, we treat events as coming from system/trusted source
			// Use nil pubkey and empty remote to indicate system-level import
			allowed, policyErr := policyManager.CheckPolicy("write", ev, nil, "")
			if policyErr != nil {
				ev.Free()
				policyRejected++
				// Only log policy errors if they're frequent (not spam)
				if policyRejected <= 10 || policyRejected%100 == 0 {
					log.W.F("[HTTP API IMPORT] Policy check failed for event %x: %v", ev.ID, policyErr)
				}
				continue
			}
			if !allowed {
				ev.Free()
				policyRejected++
				// Don't log individual policy rejections (too noisy)
				continue
			}
		}

		// Apply rate limiting before write operation if limiter is configured
		if d.rateLimiter != nil && d.rateLimiter.IsEnabled() {
			d.rateLimiter.Wait(ctx, WriteOpType)
		}

		if _, err := d.SaveEvent(ctx, ev); err != nil {
			// return the pooled buffer on error paths too
			ev.Free()
			saveErrors++
			// Only log save errors if they're frequent (not spam)
			if saveErrors <= 10 || saveErrors%100 == 0 {
				log.W.F("[HTTP API IMPORT] Failed to save event #%d: %v", saveErrors, err)
			}
			continue
		}

		// return the pooled buffer after successful save
		ev.Free()
		b = nil
		count++

		// Progress logging every logInterval with detailed information
		if time.Since(lastLogTime) >= logInterval {
			elapsed := time.Since(startTime)
			eventsPerSec := float64(count) / elapsed.Seconds()
			mbRead := float64(bytesRead) / 1024 / 1024
			mbPerSec := mbRead / elapsed.Seconds()

			// Calculate progress percentage if we have total bytes
			progressPercent := ""
			eta := ""
			if totalBytes > 0 {
				percent := float64(bytesReadTotal) / float64(totalBytes) * 100
				if percent > 100 {
					percent = 100
				}
				progressPercent = fmt.Sprintf(" | Progress: %.1f%%", percent)

				// Estimate time remaining
				if percent > 0 && percent < 100 && eventsPerSec > 0 {
					remainingBytes := totalBytes - bytesReadTotal
					remainingMB := float64(remainingBytes) / 1024 / 1024
					if mbPerSec > 0 {
						remainingSeconds := remainingMB / mbPerSec
						eta = fmt.Sprintf(" | ETA: ~%v", time.Duration(remainingSeconds)*time.Second)
					}
				}
			}

			log.I.F("[HTTP API IMPORT] Progress: %d events saved | %.2f MB read | %.0f events/sec | %.2f MB/sec%s%s",
				count, mbRead, eventsPerSec, mbPerSec, progressPercent, eta)

			// Show error summary if there are errors
			if unmarshalErrors > 0 || saveErrors > 0 || policyRejected > 0 {
				log.W.F("[HTTP API IMPORT] Errors so far: %d unmarshal, %d save, %d policy rejected",
					unmarshalErrors, saveErrors, policyRejected)
			}

			lastLogTime = time.Now()
			debug.FreeOSMemory()
		}
	}

	// Final summary with actionable information
	elapsed := time.Since(startTime)
	eventsPerSec := float64(count) / elapsed.Seconds()
	mbRead := float64(bytesRead) / 1024 / 1024
	mbPerSec := mbRead / elapsed.Seconds()

	log.I.F("[HTTP API IMPORT] ================================================")
	log.I.F("[HTTP API IMPORT] Processing complete")
	log.I.F("[HTTP API IMPORT] Events saved: %d | Data processed: %.2f MB | Time: %v",
		count, mbRead, elapsed.Round(time.Millisecond))
	log.I.F("[HTTP API IMPORT] Performance: %.0f events/sec | %.2f MB/sec", eventsPerSec, mbPerSec)

	// Show statistics
	if kind3Count > 0 || largeEventCount > 0 {
		log.I.F("[HTTP API IMPORT] Event types: %d kind-3 (follow lists), %d large events (>100KB)",
			kind3Count, largeEventCount)
	}

	// Show error summary with actionable information
	hasErrors := false
	if unmarshalErrors > 0 {
		log.W.F("[HTTP API IMPORT] WARNING: %d events failed to unmarshal", unmarshalErrors)
		log.W.F("[HTTP API IMPORT] Action: Check if file format is valid JSONL (one JSON object per line)")
		hasErrors = true
	}
	if saveErrors > 0 {
		log.W.F("[HTTP API IMPORT] WARNING: %d events failed to save", saveErrors)
		log.W.F("[HTTP API IMPORT] Action: Check database disk space and permissions")
		hasErrors = true
	}
	if policyRejected > 0 {
		log.W.F("[HTTP API IMPORT] WARNING: %d events rejected by policy", policyRejected)
		log.W.F("[HTTP API IMPORT] Action: Review ACL/policy settings if this is unexpected")
		hasErrors = true
	}
	if skipped > 0 {
		log.I.F("[HTTP API IMPORT] Info: %d empty lines skipped (this is normal)", skipped)
	}

	if !hasErrors {
		log.I.F("[HTTP API IMPORT] Result: All events imported successfully")
		log.I.F("[HTTP API IMPORT] Action: Events are now available on the relay")
	} else {
		log.W.F("[HTTP API IMPORT] Result: Import completed with errors (see warnings above)")
		log.W.F("[HTTP API IMPORT] Action: Review error messages and take appropriate action")
	}
	log.I.F("[HTTP API IMPORT] ================================================")

	if err := scan.Err(); err != nil {
		log.E.F("[HTTP API IMPORT] ERROR: Scanner error: %v", err)
		log.E.F("[HTTP API IMPORT] Action: Check file integrity and format")
		return err
	}

	return nil
}