easyai-ai-gateway/apps/api/internal/clients/topaz.go

package clients

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"time"
)

const (
	topazDefaultPollInterval = 15 * time.Second
	topazDefaultPollTimeout  = 60 * time.Minute
	topazDefaultMaxInput     = int64(2 << 30)
)

type TopazClient struct {
	HTTPClient *http.Client
	Probe      func(context.Context, string) (TopazVideoMetadata, error)
}

type TopazVideoMetadata struct {
	Width      int
	Height     int
	Duration   float64
	FrameRate  float64
	FrameCount int
	HasAudio   bool
}

type topazSource struct {
	Path       string
	Container  string
	Size       int64
	MD5        string
	Resolution map[string]any
	Duration   float64
	FrameRate  float64
	FrameCount int
}

func (c TopazClient) Run(ctx context.Context, request Request) (Response, error) {
	startedAt := time.Now()
	apiKey := credential(request.Candidate.Credentials, "apiKey", "api_key", "key", "token")
	if apiKey == "" {
		return Response{}, &ClientError{Code: "missing_credentials", Message: "Topaz API key is required", Retryable: false}
	}
	requestID := strings.TrimSpace(request.RemoteTaskID)
	payload := cloneMapAny(request.RemoteTaskPayload)
	var source *topazSource
	var target map[string]int
	if requestID == "" || strings.TrimSpace(firstNonEmptyString(payload["phase"])) != "uploaded" {
		videoURL := strings.TrimSpace(firstNonEmptyString(request.Body["video_url"], request.Body["videoUrl"]))
		if videoURL == "" {
			return Response{}, &ClientError{Code: "invalid_parameter", Message: "video_url is required", Param: "video_url", StatusCode: http.StatusBadRequest}
		}
		prepared, err := c.prepareSource(ctx, request, videoURL)
		if err != nil {
			return Response{}, err
		}
		source = &prepared
		defer os.Remove(prepared.Path)
		target = topazTargetResolution(request.Body, prepared.Resolution)
		if requestID == "" {
			created, err := c.createRequest(ctx, request, apiKey, prepared, target)
			if err != nil {
				return Response{}, annotateResponseError(err, "", startedAt, time.Now())
			}
			requestID = strings.TrimSpace(firstNonEmptyString(created["requestId"], created["request_id"], created["id"]))
			if requestID == "" {
				return Response{}, &ClientError{Code: "invalid_response", Message: "Topaz create response is missing requestId", Retryable: false}
			}
			payload = map[string]any{"phase": "created", "targetResolution": target}
			if request.OnRemoteTaskSubmitted != nil {
				if err := request.OnRemoteTaskSubmitted(requestID, payload); err != nil {
					return Response{}, err
				}
			}
		}
		if err := c.uploadSource(ctx, request, apiKey, requestID, prepared); err != nil {
			return Response{}, annotateResponseError(err, requestID, startedAt, time.Now())
		}
		payload = map[string]any{"phase": "uploaded", "targetResolution": target}
		if request.OnRemoteTaskPolled != nil {
			if err := request.OnRemoteTaskPolled(requestID, payload); err != nil {
				return Response{}, err
			}
		}
	}
	if target == nil {
		target = topazTargetFromPayload(payload, request.Body)
	}
	completed, err := c.poll(ctx, request, apiKey, requestID, target)
	if err != nil {
		return Response{}, annotateResponseError(err, requestID, startedAt, time.Now())
	}
	outputURL := topazDownloadURL(completed)
	if outputURL == "" {
		return Response{}, &ClientError{Code: "invalid_response", Message: "Topaz completed without a download URL", RequestID: requestID, Retryable: false}
	}
	metadata, err := c.probe(ctx, outputURL)
	if err != nil {
		return Response{}, &ClientError{Code: "invalid_response", Message: "cannot validate Topaz output metadata: " + err.Error(), RequestID: requestID, Retryable: true}
	}
	if target["width"] > 0 && target["height"] > 0 && (metadata.Width+4 < target["width"] || metadata.Height+4 < target["height"]) {
		return Response{}, &ClientError{Code: "invalid_response", Message: fmt.Sprintf("Topaz output resolution %dx%d does not reach target %dx%d", metadata.Width, metadata.Height, target["width"], target["height"]), RequestID: requestID, Retryable: false}
	}
	finishedAt := time.Now()
	resultItem := map[string]any{
		"type":              "video",
		"url":               outputURL,
		"video_url":         outputURL,
		"width":             metadata.Width,
		"height":            metadata.Height,
		"target_resolution": fmt.Sprintf("%dx%d", target["width"], target["height"]),
		"duration":          metadata.Duration,
		"target_frame_rate": metadata.FrameRate,
		"slow_motion_rate":  numericValue(request.Body["slow_motion_rate"], 1),
	}
	if source != nil {
		resultItem["source_resolution"] = fmt.Sprintf("%vx%v", source.Resolution["width"], source.Resolution["height"])
		resultItem["source_frame_rate"] = source.FrameRate
		resultItem["duration"] = source.Duration
	}
	return Response{
		Result: map[string]any{
			"status":           "success",
			"model":            request.Model,
			"task_id":          requestID,
			"upstream_task_id": requestID,
			"data":             []any{resultItem},
		},
		RequestID:          requestID,
		Progress:           append(providerProgress(request), Progress{Phase: "polling", Progress: 0.9, Message: "Topaz upscale completed", Payload: map[string]any{"upstreamTaskId": requestID}}),
		ResponseStartedAt:  startedAt,
		ResponseFinishedAt: finishedAt,
		ResponseDurationMS: responseDurationMS(startedAt, finishedAt),
	}, nil
}

func (c TopazClient) prepareSource(ctx context.Context, request Request, rawURL string) (topazSource, error) {
	parsed, err := url.Parse(rawURL)
	if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") {
		return topazSource{}, &ClientError{Code: "invalid_parameter", Message: "video_url must be an http(s) URL", Param: "video_url", StatusCode: http.StatusBadRequest}
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
	if err != nil {
		return topazSource{}, err
	}
	downloadClient := topazSourceHTTPClient(httpClient(request.HTTPClient, c.HTTPClient), boolishDefault(request.Candidate.PlatformConfig["allowPrivateSourceDownloads"], false))
	resp, err := downloadClient.Do(req)
	if err != nil {
		return topazSource{}, &ClientError{Code: "network", Message: err.Error(), Retryable: true}
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return topazSource{}, &ClientError{Code: "invalid_parameter", Message: "video_url download failed: " + resp.Status, Param: "video_url", StatusCode: http.StatusBadRequest}
	}
	limit := int64(numericValue(firstPresent(request.Candidate.PlatformConfig["maxInputBytes"], request.Candidate.PlatformConfig["max_input_bytes"]), float64(topazDefaultMaxInput)))
	if limit <= 0 {
		limit = topazDefaultMaxInput
	}
	tmp, err := os.CreateTemp("", "easyai-topaz-*"+topazContainerExtension(parsed.Path))
	if err != nil {
		return topazSource{}, err
	}
	path := tmp.Name()
	cleanup := func() {
		_ = tmp.Close()
		_ = os.Remove(path)
	}
	hash := md5.New()
	written, err := io.Copy(io.MultiWriter(tmp, hash), io.LimitReader(resp.Body, limit+1))
	closeErr := tmp.Close()
	if err != nil || closeErr != nil || written <= 0 || written > limit {
		cleanup()
		if written > limit {
			return topazSource{}, &ClientError{Code: "invalid_parameter", Message: "video source exceeds configured size limit", Param: "video_url", StatusCode: http.StatusBadRequest}
		}
		return topazSource{}, firstError(err, closeErr)
	}
	metadata, err := c.probe(ctx, path)
	if err != nil || metadata.Width <= 0 || metadata.Height <= 0 {
		cleanup()
		return topazSource{}, &ClientError{Code: "invalid_parameter", Message: "cannot probe source video metadata", Param: "video_url", StatusCode: http.StatusBadRequest}
	}
	frameRate := metadata.FrameRate
	if frameRate <= 0 {
		frameRate = numericValue(request.Candidate.PlatformConfig["frameRate"], 30)
	}
	duration := math.Max(1, metadata.Duration)
	frameCount := metadata.FrameCount
	if frameCount <= 0 {
		frameCount = int(math.Round(duration * frameRate))
	}
	return topazSource{
		Path: path, Container: strings.TrimPrefix(strings.ToLower(filepath.Ext(path)), "."), Size: written,
		MD5: hex.EncodeToString(hash.Sum(nil)), Resolution: map[string]any{"width": metadata.Width, "height": metadata.Height},
		Duration: duration, FrameRate: frameRate, FrameCount: frameCount,
	}, nil
}

func (c TopazClient) createRequest(ctx context.Context, request Request, apiKey string, source topazSource, target map[string]int) (map[string]any, error) {
	model := topazModelName(firstNonEmptyString(request.Candidate.ProviderModelName, request.Model))
	scale := math.Max(float64(target["width"])/numericValue(source.Resolution["width"], 1), float64(target["height"])/numericValue(source.Resolution["height"], 1))
	filter := map[string]any{"model": model}
	if scale > 1 {
		key := "scale"
		if model == "slf-2" || model == "slhq-1" || model == "slm-1" || model == "slp-2.5" {
			key = "upscaling_factor"
		}
		filter[key] = math.Round(scale*1000) / 1000
	}
	preserveAudio := boolishDefault(request.Body["preserve_audio"], true)
	filters := []any{filter}
	targetFrameRate := numericValue(firstPresent(request.Body["target_frame_rate"], request.Body["output_frame_rate"]), source.FrameRate)
	slowMotionRate := numericValue(request.Body["slow_motion_rate"], 1)
	if targetFrameRate <= 0 {
		targetFrameRate = source.FrameRate
	}
	if slowMotionRate <= 0 {
		slowMotionRate = 1
	}
	if targetFrameRate != source.FrameRate || slowMotionRate > 1 {
		interpolationModel := strings.TrimSpace(firstNonEmptyString(request.Body["frame_interpolation_model"]))
		if interpolationModel == "" {
			return nil, &ClientError{Code: "invalid_parameter", Message: "frame_interpolation_model is required when output frame rate or slow motion changes", Param: "frame_interpolation_model", StatusCode: http.StatusBadRequest}
		}
		interpolation := map[string]any{"model": topazModelName(interpolationModel)}
		if targetFrameRate > 0 {
			interpolation["fps"] = targetFrameRate
		}
		if slowMotionRate > 1 {
			interpolation["slowmo"] = slowMotionRate
		}
		filters = append(filters, interpolation)
	}
	body := map[string]any{
		"source":  map[string]any{"container": source.Container, "size": source.Size, "duration": source.Duration, "frameCount": source.FrameCount, "frameRate": source.FrameRate, "resolution": source.Resolution},
		"filters": filters,
		"output":  map[string]any{"resolution": target, "frameRate": targetFrameRate, "audioCodec": "AAC", "audioTransfer": map[bool]string{true: "Copy", false: "None"}[preserveAudio], "dynamicCompressionLevel": "High", "videoEncoder": "H265", "videoProfile": "Main", "container": "mp4"},
	}
	return c.topazJSON(ctx, request, apiKey, http.MethodPost, "/video/", body)
}

func (c TopazClient) uploadSource(ctx context.Context, request Request, apiKey, requestID string, source topazSource) error {
	accepted, err := c.topazJSON(ctx, request, apiKey, http.MethodPatch, "/video/"+url.PathEscape(requestID)+"/accept", nil)
	if err != nil {
		return err
	}
	urls := stringList(accepted["urls"])
	if len(urls) == 0 {
		return &ClientError{Code: "invalid_response", Message: "Topaz accept response has no upload URLs", RequestID: requestID}
	}
	file, err := os.Open(source.Path)
	if err != nil {
		return err
	}
	defer file.Close()
	chunkSize := (source.Size + int64(len(urls)) - 1) / int64(len(urls))
	results := make([]any, 0, len(urls))
	for index, uploadURL := range urls {
		remaining := source.Size - int64(index)*chunkSize
		if remaining <= 0 {
			break
		}
		length := minInt64(chunkSize, remaining)
		section := io.NewSectionReader(file, int64(index)*chunkSize, length)
		req, err := http.NewRequestWithContext(ctx, http.MethodPut, uploadURL, section)
		if err != nil {
			return err
		}
		req.ContentLength = length
		req.Header.Set("Content-Type", topazContainerContentType(source.Container))
		resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req)
		if err != nil {
			return &ClientError{Code: "network", Message: err.Error(), Retryable: true}
		}
		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<20))
		resp.Body.Close()
		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
			return &ClientError{Code: "provider_failed", Message: "Topaz part upload failed: " + resp.Status, StatusCode: resp.StatusCode, Retryable: resp.StatusCode >= 500}
		}
		etag := strings.Trim(resp.Header.Get("ETag"), `"`)
		if etag == "" {
			return &ClientError{Code: "invalid_response", Message: "Topaz part upload is missing ETag", Retryable: false}
		}
		results = append(results, map[string]any{"partNum": index + 1, "eTag": etag})
	}
	_, err = c.topazJSON(ctx, request, apiKey, http.MethodPatch, "/video/"+url.PathEscape(requestID)+"/complete-upload/", map[string]any{"md5Hash": source.MD5, "uploadResults": results})
	return err
}

func (c TopazClient) poll(ctx context.Context, request Request, apiKey, requestID string, target map[string]int) (map[string]any, error) {
	interval := universalDurationConfig(request.Candidate.PlatformConfig, topazDefaultPollInterval, "pollIntervalMs", "poll_interval_ms")
	timeout := universalDurationConfig(request.Candidate.PlatformConfig, topazDefaultPollTimeout, "pollTimeoutMs", "poll_timeout_ms", "timeoutMs")
	deadline := time.NewTimer(timeout)
	defer deadline.Stop()
	for {
		status, err := c.topazJSON(ctx, request, apiKey, http.MethodGet, "/video/"+url.PathEscape(requestID)+"/status", nil)
		if err != nil {
			return nil, err
		}
		state := strings.ToLower(strings.TrimSpace(firstNonEmptyString(status["status"], status["state"])))
		if request.OnRemoteTaskPolled != nil {
			if err := request.OnRemoteTaskPolled(requestID, map[string]any{"phase": "uploaded", "status": state, "targetResolution": target}); err != nil {
				return nil, err
			}
		}
		switch state {
		case "complete", "completed", "success", "succeeded":
			return status, nil
		case "failed", "failure", "cancelled", "canceled", "error":
			return nil, &ClientError{Code: "provider_failed", Message: firstNonEmptyString(status["message"], "Topaz task failed"), RequestID: requestID, Retryable: false}
		}
		select {
		case <-ctx.Done():
			return nil, &ClientError{Code: "cancelled", Message: ctx.Err().Error(), RequestID: requestID, Retryable: true}
		case <-deadline.C:
			return nil, &ClientError{Code: "timeout", Message: "Topaz task polling timed out", RequestID: requestID, Retryable: true}
		case <-time.After(interval):
		}
	}
}

func (c TopazClient) topazJSON(ctx context.Context, request Request, apiKey, method, path string, body map[string]any) (map[string]any, error) {
	var reader io.Reader
	if body != nil {
		raw, _ := json.Marshal(body)
		reader = bytes.NewReader(raw)
	}
	req, err := http.NewRequestWithContext(ctx, method, providerURL(request.Candidate.BaseURL, path), reader)
	if err != nil {
		return nil, err
	}
	req.Header.Set("X-API-Key", apiKey)
	req.Header.Set("Accept", "application/json")
	if body != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req)
	if err != nil {
		return nil, &ClientError{Code: "network", Message: err.Error(), Retryable: true}
	}
	result, decodeErr := decodeHTTPResponse(resp)
	if decodeErr != nil {
		return nil, decodeErr
	}
	return result, nil
}

func (c TopazClient) probe(ctx context.Context, target string) (TopazVideoMetadata, error) {
	if c.Probe != nil {
		return c.Probe(ctx, target)
	}
	return probeTopazVideo(ctx, target)
}

func probeTopazVideo(ctx context.Context, target string) (TopazVideoMetadata, error) {
	if _, err := exec.LookPath("ffprobe"); err != nil {
		return TopazVideoMetadata{}, err
	}
	probeCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	output, err := exec.CommandContext(probeCtx, "ffprobe", "-v", "error", "-show_entries", "format=duration:stream=codec_type,width,height,avg_frame_rate,nb_frames", "-of", "json", target).Output()
	if err != nil {
		return TopazVideoMetadata{}, err
	}
	var decoded struct {
		Format struct {
			Duration string `json:"duration"`
		} `json:"format"`
		Streams []struct {
			CodecType   string `json:"codec_type"`
			Width       int    `json:"width"`
			Height      int    `json:"height"`
			AverageRate string `json:"avg_frame_rate"`
			FrameCount  string `json:"nb_frames"`
		} `json:"streams"`
	}
	if err := json.Unmarshal(output, &decoded); err != nil {
		return TopazVideoMetadata{}, err
	}
	metadata := TopazVideoMetadata{}
	metadata.Duration, _ = strconv.ParseFloat(decoded.Format.Duration, 64)
	for _, stream := range decoded.Streams {
		if stream.CodecType == "audio" {
			metadata.HasAudio = true
		}
		if stream.CodecType != "video" || stream.Width <= 0 || stream.Height <= 0 {
			continue
		}
		metadata.Width, metadata.Height = stream.Width, stream.Height
		metadata.FrameRate = parseTopazRate(stream.AverageRate)
		metadata.FrameCount, _ = strconv.Atoi(stream.FrameCount)
	}
	if metadata.Width <= 0 || metadata.Height <= 0 {
		return TopazVideoMetadata{}, errors.New("video stream metadata is missing")
	}
	return metadata, nil
}

func topazTargetResolution(body map[string]any, source map[string]any) map[string]int {
	width := int(numericValue(body["output_width"], 0))
	height := int(numericValue(body["output_height"], 0))
	if width > 0 && height > 0 {
		return map[string]int{"width": width, "height": height}
	}
	value := strings.ToLower(strings.TrimSpace(firstNonEmptyString(body["target_resolution"], body["output_resolution"], "1080p")))
	if parsedWidth, parsedHeight, ok := parseTopazSize(value); ok {
		return map[string]int{"width": parsedWidth, "height": parsedHeight}
	}
	base := map[string][2]int{"480p": {854, 480}, "720p": {1280, 720}, "1080p": {1920, 1080}, "1440p": {2560, 1440}, "2k": {2560, 1440}, "2160p": {3840, 2160}, "4k": {3840, 2160}}
	value = strings.TrimSuffix(value, "_upscale")
	target, ok := base[value]
	if !ok {
		target = base["1080p"]
	}
	sourceWidth := numericValue(source["width"], 1)
	sourceHeight := numericValue(source["height"], 1)
	if sourceWidth == sourceHeight {
		side := minInt(target[0], target[1])
		return map[string]int{"width": side, "height": side}
	}
	if sourceWidth > sourceHeight {
		return map[string]int{"width": int(math.Round(sourceWidth / sourceHeight * float64(target[1]))), "height": target[1]}
	}
	return map[string]int{"width": target[1], "height": int(math.Round(sourceHeight / sourceWidth * float64(target[1])))}
}

func topazTargetFromPayload(payload map[string]any, body map[string]any) map[string]int {
	if raw, ok := payload["targetResolution"].(map[string]any); ok {
		return map[string]int{"width": int(numericValue(raw["width"], 0)), "height": int(numericValue(raw["height"], 0))}
	}
	return topazTargetResolution(body, map[string]any{"width": 16, "height": 9})
}

func topazDownloadURL(result map[string]any) string {
	if download, ok := result["download"].(map[string]any); ok {
		return firstNonEmptyString(download["url"], download["downloadUrl"])
	}
	return firstNonEmptyString(result["download_url"], result["downloadUrl"], result["url"])
}

func topazModelName(model string) string {
	switch strings.TrimSpace(model) {
	case "easy-proteus-standard-4":
		return "prob-4"
	case "easy-starlight-fast-2":
		return "slf-2"
	case "easy-starlight-hq-1":
		return "slhq-1"
	case "easy-starlight-mini-1":
		return "slm-1"
	default:
		return strings.TrimSpace(model)
	}
}

func topazContainerExtension(path string) string {
	ext := strings.ToLower(filepath.Ext(strings.Split(path, "?")[0]))
	switch ext {
	case ".mov", ".mkv", ".webm", ".mp4":
		return ext
	default:
		return ".mp4"
	}
}

func topazContainerContentType(container string) string {
	switch strings.ToLower(container) {
	case "mov":
		return "video/quicktime"
	case "mkv":
		return "video/x-matroska"
	case "webm":
		return "video/webm"
	default:
		return "video/mp4"
	}
}

func parseTopazRate(value string) float64 {
	parts := strings.Split(value, "/")
	if len(parts) == 2 {
		numerator, _ := strconv.ParseFloat(parts[0], 64)
		denominator, _ := strconv.ParseFloat(parts[1], 64)
		if denominator > 0 {
			return numerator / denominator
		}
	}
	parsed, _ := strconv.ParseFloat(value, 64)
	return parsed
}

func parseTopazSize(value string) (int, int, bool) {
	parts := strings.Split(strings.ReplaceAll(value, " ", ""), "x")
	if len(parts) != 2 {
		return 0, 0, false
	}
	width, errWidth := strconv.Atoi(parts[0])
	height, errHeight := strconv.Atoi(parts[1])
	return width, height, errWidth == nil && errHeight == nil && width > 0 && height > 0
}

func stringList(value any) []string {
	items, ok := value.([]any)
	if !ok {
		if values, ok := value.([]string); ok {
			return values
		}
		return nil
	}
	out := make([]string, 0, len(items))
	for _, item := range items {
		if text := strings.TrimSpace(fmt.Sprint(item)); text != "" {
			out = append(out, text)
		}
	}
	return out
}

func boolishDefault(value any, fallback bool) bool {
	if value == nil {
		return fallback
	}
	switch typed := value.(type) {
	case bool:
		return typed
	case string:
		parsed, err := strconv.ParseBool(strings.TrimSpace(typed))
		if err == nil {
			return parsed
		}
	}
	return fallback
}

func firstError(values ...error) error {
	for _, value := range values {
		if value != nil {
			return value
		}
	}
	return nil
}

func minInt64(left, right int64) int64 {
	if left < right {
		return left
	}
	return right
}

func minInt(left, right int) int {
	if left < right {
		return left
	}
	return right
}

func topazSourceHTTPClient(base *http.Client, allowPrivate bool) *http.Client {
	client := *base
	client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }
	if client.Timeout <= 0 {
		client.Timeout = 10 * time.Minute
	}
	if allowPrivate {
		return &client
	}
	transport := http.DefaultTransport.(*http.Transport).Clone()
	transport.Proxy = nil
	transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) {
		host, port, err := net.SplitHostPort(address)
		if err != nil {
			return nil, err
		}
		addresses, err := net.DefaultResolver.LookupIPAddr(ctx, host)
		if err != nil || len(addresses) == 0 {
			return nil, errors.New("video source DNS resolution failed")
		}
		for _, address := range addresses {
			if topazBlockedAddress(address.IP) {
				return nil, errors.New("video source resolved to a blocked network")
			}
		}
		dialer := &net.Dialer{Timeout: 10 * time.Second}
		var attempts []error
		for _, resolved := range addresses {
			connection, dialErr := dialer.DialContext(ctx, network, net.JoinHostPort(resolved.IP.String(), port))
			if dialErr == nil {
				return connection, nil
			}
			attempts = append(attempts, dialErr)
		}
		return nil, errors.Join(attempts...)
	}
	client.Transport = transport
	return &client
}

func topazBlockedAddress(ip net.IP) bool {
	return ip == nil || ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsUnspecified() || ip.IsMulticast()
}