// Mount mounts a fuse fs.FS at a given location, and returns a Mount instance.
// parent is a ContextGroup to bind the mount's ContextGroup to.
func NewMount(p goprocess.Process, fsys fs.FS, mountpoint string, allow_other bool) (Mount, error) {
	var conn *fuse.Conn
	var err error

	if allow_other {
		conn, err = fuse.Mount(mountpoint, fuse.AllowOther())
	} else {
		conn, err = fuse.Mount(mountpoint)
	}

	if err != nil {
		return nil, err
	}

	m := &mount{
		mpoint:   mountpoint,
		fuseConn: conn,
		filesys:  fsys,
		proc:     goprocess.WithParent(p), // link it to parent.
	}
	m.proc.SetTeardown(m.unmount)

	// launch the mounting process.
	if err := m.mount(); err != nil {
		m.Unmount() // just in case.
		return nil, err
	}

	return m, nil
}

func newNAT(realNAT nat.NAT) *NAT {
	return &NAT{
		nat:      realNAT,
		proc:     goprocess.WithParent(goprocess.Background()),
		mappings: make(map[*mapping]struct{}),
	}
}

func NewWorker(e exchange.Interface, c Config) *Worker {
	if c.NumWorkers < 1 {
		c.NumWorkers = 1 // provide a sane default
	}
	w := &Worker{
		exchange: e,
		added:    make(chan *blocks.Block, c.ClientBufferSize),
		process:  process.WithParent(process.Background()), // internal management
	}
	w.start(c)
	return w
}

func newQueryRunner(q *dhtQuery) *dhtQueryRunner {
	proc := process.WithParent(process.Background())
	ctx := ctxproc.WithProcessClosing(context.Background(), proc)
	return &dhtQueryRunner{
		query:          q,
		peersToQuery:   queue.NewChanQueue(ctx, queue.NewXORDistancePQ(q.key)),
		peersRemaining: todoctr.NewSyncCounter(),
		peersSeen:      pset.New(),
		rateLimit:      make(chan struct{}, q.concurrency),
		proc:           proc,
	}
}

// WithContext constructs and returns a Process that respects
// given context. It is the equivalent of:
//
//   func ProcessWithContext(ctx context.Context) goprocess.Process {
//     p := goprocess.WithParent(goprocess.Background())
//     go func() {
//       <-ctx.Done()
//       p.Close()
//     }()
//     return p
//   }
//
func WithContext(ctx context.Context) goprocess.Process {
	if ctx == nil {
		panic("nil Context")
	}

	p := goprocess.WithParent(goprocess.Background())
	go func() {
		<-ctx.Done()
		p.Close()
	}()
	return p
}

func newNatManager(host *BasicHost) *natManager {
	nmgr := &natManager{
		host:  host,
		ready: make(chan struct{}),
		proc:  goprocess.WithParent(host.proc),
	}

	// teardown
	nmgr.proc = goprocess.WithTeardown(func() error {
		// on closing, unregister from network notifications.
		host.Network().StopNotify((*nmgrNetNotifiee)(nmgr))
		return nil
	})

	// host is our parent. close when host closes.
	host.proc.AddChild(nmgr.proc)

	// discover the nat.
	nmgr.discoverNAT()
	return nmgr
}

func run(ipfsPath, watchPath string) error {

	proc := process.WithParent(process.Background())
	log.Printf("running IPFSWatch on '%s' using repo at '%s'...", watchPath, ipfsPath)

	ipfsPath, err := homedir.Expand(ipfsPath)
	if err != nil {
		return err
	}
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		return err
	}
	defer watcher.Close()

	if err := addTree(watcher, watchPath); err != nil {
		return err
	}

	r, err := fsrepo.Open(ipfsPath)
	if err != nil {
		// TODO handle case: daemon running
		// TODO handle case: repo doesn't exist or isn't initialized
		return err
	}
	node, err := core.NewIPFSNode(context.Background(), core.Online(r))
	if err != nil {
		return err
	}
	defer node.Close()

	if *http {
		addr := "/ip4/127.0.0.1/tcp/5001"
		var opts = []corehttp.ServeOption{
			corehttp.GatewayOption(true),
			corehttp.WebUIOption,
			corehttp.CommandsOption(cmdCtx(node, ipfsPath)),
		}
		proc.Go(func(p process.Process) {
			if err := corehttp.ListenAndServe(node, addr, opts...); err != nil {
				return
			}
		})
	}

	interrupts := make(chan os.Signal)
	signal.Notify(interrupts, os.Interrupt, os.Kill)

	for {
		select {
		case <-interrupts:
			return nil
		case e := <-watcher.Events:
			log.Printf("received event: %s", e)
			isDir, err := IsDirectory(e.Name)
			if err != nil {
				continue
			}
			switch e.Op {
			case fsnotify.Remove:
				if isDir {
					if err := watcher.Remove(e.Name); err != nil {
						return err
					}
				}
			default:
				// all events except for Remove result in an IPFS.Add, but only
				// directory creation triggers a new watch
				switch e.Op {
				case fsnotify.Create:
					if isDir {
						addTree(watcher, e.Name)
					}
				}
				proc.Go(func(p process.Process) {
					file, err := os.Open(e.Name)
					if err != nil {
						log.Println(err)
					}
					defer file.Close()
					k, err := coreunix.Add(node, file)
					if err != nil {
						log.Println(err)
					}
					log.Printf("added %s... key: %s", e.Name, k)
				})
			}
		case err := <-watcher.Errors:
			log.Println(err)
		}
	}
	return nil
}

// WithContext constructs and returns a Process that respects
// given context. It is the equivalent of:
//
//   func ProcessWithContext(ctx context.Context) goprocess.Process {
//     p := goprocess.WithParent(goprocess.Background())
//     CloseAfterContext(p, ctx)
//     return p
//   }
//
func WithContext(ctx context.Context) goprocess.Process {
	p := goprocess.WithParent(goprocess.Background())
	CloseAfterContext(p, ctx)
	return p
}

func NewRateLimiter(parent process.Process, limit int) *RateLimiter {
	proc := process.WithParent(parent)
	return &RateLimiter{Process: proc, limiter: LimitChan(limit)}
}