code:
{- git-annex assistant commit thread
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

{-# LANGUAGE CPP #-}

module Assistant.Threads.Committer where

import Assistant.Common
import Assistant.Changes
import Assistant.Types.Changes
import Assistant.Commits
import Assistant.Alert
import Assistant.DaemonStatus
import Assistant.TransferQueue
import Assistant.Drop
import Logs.Transfer
import Logs.Location
import qualified Annex.Queue
import qualified Git.LsFiles
import qualified Command.Add
import Utility.ThreadScheduler
import qualified Utility.Lsof as Lsof
import qualified Utility.DirWatcher as DirWatcher
import Types.KeySource
import Config
import Annex.Exception
import Annex.Content
import Annex.Link
import Annex.CatFile
import qualified Annex
import Utility.InodeCache
import Annex.Content.Direct
import qualified Command.Sync
import qualified Git.Branch

import Data.Time.Clock
import Data.Tuple.Utils
import qualified Data.Set as S
import qualified Data.Map as M
import Data.Either
import Control.Concurrent

{- This thread makes git commits at appropriate times. -}
commitThread :: NamedThread
commitThread = namedThread "Committer" $ do
	havelsof <- liftIO $ inPath "lsof"
	delayadd <- liftAnnex $
		maybe delayaddDefault (return . Just . Seconds)
			=<< annexDelayAdd <$> Annex.getGitConfig
	waitChangeTime $ \(changes, time) -> do
		readychanges <- handleAdds havelsof delayadd changes
		if shouldCommit False time (length readychanges) readychanges
			then do
				debug
					[ "committing"
					, show (length readychanges)
					, "changes"
					]
				void $ alertWhile commitAlert $
					liftAnnex commitStaged
				recordCommit
				let numchanges = length readychanges
				mapM_ checkChangeContent readychanges
				return numchanges
			else do
				refill readychanges
				return 0

refill :: [Change] -> Assistant ()
refill [] = noop
refill cs = do
	debug ["delaying commit of", show (length cs), "changes"]
	refillChanges cs

{- Wait for one or more changes to arrive to be committed, and then
 - runs an action to commit them. If more changes arrive while this is
 - going on, they're handled intelligently, batching up changes into
 - large commits where possible, doing rename detection, and
 - commiting immediately otherwise. -}
waitChangeTime :: (([Change], UTCTime) -> Assistant Int) -> Assistant ()
waitChangeTime a = waitchanges 0
  where
	waitchanges lastcommitsize = do
		-- Wait one one second as a simple rate limiter.
		liftIO $ threadDelaySeconds (Seconds 1)
		-- Now, wait until at least one change is available for
		-- processing.
		cs <- getChanges
		handlechanges cs lastcommitsize
	handlechanges changes lastcommitsize = do
		let len = length changes
		-- See if now's a good time to commit.
		now <- liftIO getCurrentTime
		scanning <- not . scanComplete <$> getDaemonStatus
		case (lastcommitsize >= maxCommitSize, shouldCommit scanning now len changes, possiblyrename changes) of
			(True, True, _)
				| len > maxCommitSize ->
					a (changes, now) >>= waitchanges
				| otherwise -> aftermaxcommit changes
			(_, True, False) ->
				a (changes, now) >>= waitchanges
			(_, True, True) -> do
				morechanges <- getrelatedchanges changes
				a (changes ++ morechanges, now) >>= waitchanges
			_ -> do
				refill changes
				waitchanges lastcommitsize

	{- Did we perhaps only get one of the AddChange and RmChange pair
	 - that make up a file rename? Or some of the pairs that make up
	 - a directory rename?
	 -}
	possiblyrename = all renamepart

	renamepart (PendingAddChange _ _) = True
	renamepart c = isRmChange c

	{- Gets changes related to the passed changes, without blocking
	 - very long.
	 -
	 - If there are multiple RmChanges, this is probably a directory
	 - rename, in which case it may be necessary to wait longer to get
	 - all the Changes involved.
	 -}
	getrelatedchanges oldchanges
		| length (filter isRmChange oldchanges) > 1 =
			concat <$> getbatchchanges []
		| otherwise = do
			liftIO humanImperceptibleDelay
			getAnyChanges
	getbatchchanges cs = do
		liftIO $ threadDelay $ fromIntegral $ oneSecond `div` 10
		cs' <- getAnyChanges
		if null cs'
			then return cs
			else getbatchchanges (cs':cs)

	{- The last commit was maximum size, so it's very likely there
	 - are more changes and we'd like to ensure we make another commit
	 - of maximum size if possible.
	 -
	 - But, it can take a while for the Watcher to wake back up
	 - after a commit. It can get blocked by another thread
	 - that is using the Annex state, such as a git-annex branch
	 - commit. Especially after such a large commit, this can
	 - take several seconds. When this happens, it defeats the
	 - normal commit batching, which sees some old changes the
	 - Watcher found while the commit was being prepared, and sees
	 - no recent ones, and wants to commit immediately.
	 -
	 - All that we need to do, then, is wait for the Watcher to
	 - wake up, and queue up one more change.
	 -
	 - However, it's also possible that we're at the end of changes for
	 - now. So to avoid waiting a really long time before committing
	 - those changes we have, poll for up to 30 seconds, and then
	 - commit them.
	 -
	 - Also, try to run something in Annex, to ensure we block
	 - longer if the Annex state is indeed blocked.
	 -}
	aftermaxcommit oldchanges = loop (30 :: Int)
	  where
	  	loop 0 = continue oldchanges
	  	loop n = do
			liftAnnex noop -- ensure Annex state is free
			liftIO $ threadDelaySeconds (Seconds 1)
			changes <- getAnyChanges
			if null changes
				then loop (n - 1)
				else continue (oldchanges ++ changes)
		continue cs
			| null cs = waitchanges 0
			| otherwise = handlechanges cs 0

isRmChange :: Change -> Bool
isRmChange (Change { changeInfo = i }) | i == RmChange = True
isRmChange _ = False

{- An amount of time that is hopefully imperceptably short for humans,
 - while long enough for a computer to get some work done.
 - Note that 0.001 is a little too short for rename change batching to
 - work. -}
humanImperceptibleInterval :: NominalDiffTime
humanImperceptibleInterval = 0.01

humanImperceptibleDelay :: IO ()
humanImperceptibleDelay = threadDelay $
	truncate $ humanImperceptibleInterval * fromIntegral oneSecond

maxCommitSize :: Int
maxCommitSize = 5000

{- Decide if now is a good time to make a commit.
 - Note that the list of changes has an undefined order.
 -
 - Current strategy: If there have been 10 changes within the past second,
 - a batch activity is taking place, so wait for later.
 -}
shouldCommit :: Bool -> UTCTime -> Int -> [Change] -> Bool
shouldCommit scanning now len changes
	| scanning = len >= maxCommitSize
	| len == 0 = False
	| len >= maxCommitSize = True
	| length recentchanges < 10 = True
	| otherwise = False -- batch activity
  where
	thissecond c = timeDelta c <= 1
	recentchanges = filter thissecond changes
	timeDelta c = now `diffUTCTime` changeTime c

commitStaged :: Annex Bool
commitStaged = do
	{- This could fail if there's another commit being made by
	 - something else. -}
	v <- tryAnnex Annex.Queue.flush
	case v of
		Left _ -> return False
		Right _ -> do
			ok <- Command.Sync.commitStaged Git.Branch.AutomaticCommit ""
			when ok $
				Command.Sync.updateSyncBranch =<< inRepo Git.Branch.current
			return ok

{- OSX needs a short delay after a file is added before locking it down,
 - when using a non-direct mode repository, as pasting a file seems to
 - try to set file permissions or otherwise access the file after closing
 - it. -}
delayaddDefault :: Annex (Maybe Seconds)
#ifdef darwin_HOST_OS
delayaddDefault = ifM isDirect
	( return Nothing
	, return $ Just $ Seconds 1
	)
#else
delayaddDefault = return Nothing
#endif

{- If there are PendingAddChanges, or InProcessAddChanges, the files
 - have not yet actually been added to the annex, and that has to be done
 - now, before committing.
 -
 - Deferring the adds to this point causes batches to be bundled together,
 - which allows faster checking with lsof that the files are not still open
 - for write by some other process, and faster checking with git-ls-files
 - that the files are not already checked into git.
 -
 - When a file is added, Inotify will notice the new symlink. So this waits
 - for additional Changes to arrive, so that the symlink has hopefully been
 - staged before returning, and will be committed immediately.
 -
 - OTOH, for kqueue, eventsCoalesce, so instead the symlink is directly
 - created and staged.
 -
 - Returns a list of all changes that are ready to be committed.
 - Any pending adds that are not ready yet are put back into the ChangeChan,
 - where they will be retried later.
 -}
handleAdds :: Bool -> Maybe Seconds -> [Change] -> Assistant [Change]
handleAdds havelsof delayadd cs = returnWhen (null incomplete) $ do
	let (pending, inprocess) = partition isPendingAddChange incomplete
	direct <- liftAnnex isDirect
	(pending', cleanup) <- if direct
		then return (pending, noop)
		else findnew pending
	(postponed, toadd) <- partitionEithers <$> safeToAdd havelsof delayadd pending' inprocess
	cleanup

	unless (null postponed) $
		refillChanges postponed

	returnWhen (null toadd) $ do
		added <- addaction toadd $
			catMaybes <$> if direct
				then adddirect toadd
				else forM toadd add
		if DirWatcher.eventsCoalesce || null added || direct
			then return $ added ++ otherchanges
			else do
				r <- handleAdds havelsof delayadd =<< getChanges
				return $ r ++ added ++ otherchanges
  where
	(incomplete, otherchanges) = partition (\c -> isPendingAddChange c || isInProcessAddChange c) cs

	findnew [] = return ([], noop)
	findnew pending@(exemplar:_) = do
		(newfiles, cleanup) <- liftAnnex $
			inRepo (Git.LsFiles.notInRepo False $ map changeFile pending)
		-- note: timestamp info is lost here
		let ts = changeTime exemplar
		return (map (PendingAddChange ts) newfiles, void $ liftIO cleanup)

	returnWhen c a
		| c = return otherchanges
		| otherwise = a

	add :: Change -> Assistant (Maybe Change)
	add change@(InProcessAddChange { keySource = ks }) =
		catchDefaultIO Nothing <~> doadd
	  where
	  	doadd = sanitycheck ks $ do
			(mkey, mcache) <- liftAnnex $ do
				showStart "add" $ keyFilename ks
				Command.Add.ingest $ Just ks
			maybe (failedingest change) (done change mcache $ keyFilename ks) mkey
	add _ = return Nothing

	{- In direct mode, avoid overhead of re-injesting a renamed
	 - file, by examining the other Changes to see if a removed
	 - file has the same InodeCache as the new file. If so,
	 - we can just update bookkeeping, and stage the file in git.
	 -}
	adddirect :: [Change] -> Assistant [Maybe Change]
	adddirect toadd = do
		ct <- liftAnnex compareInodeCachesWith
		m <- liftAnnex $ removedKeysMap ct cs
		delta <- liftAnnex getTSDelta
		if M.null m
			then forM toadd add
			else forM toadd $ \c -> do
				mcache <- liftIO $ genInodeCache (changeFile c) delta
				case mcache of
					Nothing -> add c
					Just cache ->
						case M.lookup (inodeCacheToKey ct cache) m of
							Nothing -> add c
							Just k -> fastadd c k

	fastadd :: Change -> Key -> Assistant (Maybe Change)
	fastadd change key = do
		let source = keySource change
		liftAnnex $ Command.Add.finishIngestDirect key source
		done change Nothing (keyFilename source) key

	removedKeysMap :: InodeComparisonType -> [Change] -> Annex (M.Map InodeCacheKey Key)
	removedKeysMap ct l = do
		mks <- forM (filter isRmChange l) $ \c ->
			catKeyFile $ changeFile c
		M.fromList . concat <$> mapM mkpairs (catMaybes mks)
	  where
		mkpairs k = map (\c -> (inodeCacheToKey ct c, k)) <$>
			recordedInodeCache k

	failedingest change = do
		refill [retryChange change]
		liftAnnex showEndFail
		return Nothing

	done change mcache file key = liftAnnex $ do
		logStatus key InfoPresent
		link <- ifM isDirect
			( inRepo $ gitAnnexLink file key
			, Command.Add.link file key mcache
			)
		whenM (pure DirWatcher.eventsCoalesce <||> isDirect) $
			stageSymlink file =<< hashSymlink link
		showEndOk
		return $ Just $ finishedChange change key

	{- Check that the keysource's keyFilename still exists,
	 - and is still a hard link to its contentLocation,
	 - before ingesting it. -}
	sanitycheck keysource a = do
		fs <- liftIO $ getSymbolicLinkStatus $ keyFilename keysource
		ks <- liftIO $ getSymbolicLinkStatus $ contentLocation keysource
		if deviceID ks == deviceID fs && fileID ks == fileID fs
			then a
			else do
				-- remove the hard link
				when (contentLocation keysource /= keyFilename keysource) $
					void $ liftIO $ tryIO $ removeFile $ contentLocation keysource
				return Nothing

	{- Shown an alert while performing an action to add a file or
	 - files. When only a few files are added, their names are shown
	 - in the alert. When it's a batch add, the number of files added
	 - is shown.
	 -
	 - Add errors tend to be transient and will be
	 - automatically dealt with, so the alert is always told
	 - the add succeeded.
	 -}
	addaction [] a = a
	addaction toadd a = alertWhile' (addFileAlert $ map changeFile toadd) $
		(,)
			<$> pure True
			<*> a

{- Files can Either be Right to be added now,
 - or are unsafe, and must be Left for later.
 -
 - Check by running lsof on the repository.
 -}
safeToAdd :: Bool -> Maybe Seconds -> [Change] -> [Change] -> Assistant [Either Change Change]
safeToAdd _ _ [] [] = return []
safeToAdd havelsof delayadd pending inprocess = do
	maybe noop (liftIO . threadDelaySeconds) delayadd
	liftAnnex $ do
		keysources <- forM pending $ Command.Add.lockDown . changeFile
		let inprocess' = inprocess ++ mapMaybe mkinprocess (zip pending keysources)
		openfiles <- if havelsof
			then S.fromList . map fst3 . filter openwrite <$>
				findopenfiles (map keySource inprocess')
			else pure S.empty
		let checked = map (check openfiles) inprocess'

		{- If new events are received when files are closed,
		 - there's no need to retry any changes that cannot
		 - be done now. -}
		if DirWatcher.closingTracked
			then do
				mapM_ canceladd $ lefts checked
				allRight $ rights checked
			else return checked
  where
	check openfiles change@(InProcessAddChange { keySource = ks })
		| S.member (contentLocation ks) openfiles = Left change
	check _ change = Right change

	mkinprocess (c, Just ks) = Just InProcessAddChange
		{ changeTime = changeTime c
		, keySource = ks
		}
	mkinprocess (_, Nothing) = Nothing

	canceladd (InProcessAddChange { keySource = ks }) = do
		warning $ keyFilename ks
			++ " still has writers, not adding"
		-- remove the hard link
		when (contentLocation ks /= keyFilename ks) $
			void $ liftIO $ tryIO $ removeFile $ contentLocation ks
	canceladd _ = noop

	openwrite (_file, mode, _pid)
		| mode == Lsof.OpenWriteOnly = True
		| mode == Lsof.OpenReadWrite = True
		| mode == Lsof.OpenUnknown = True
		| otherwise = False

	allRight = return . map Right

	{- Normally the KeySources are locked down inside the temp directory,
	 - so can just lsof that, which is quite efficient.
	 -
	 - In crippled filesystem mode, there is no lock down, so must run lsof
	 - on each individual file.
	 -}
	findopenfiles keysources = ifM crippledFileSystem
		( liftIO $ do
			let segments = segmentXargs $ map keyFilename keysources
			concat <$> forM segments (\fs -> Lsof.query $ "--" : fs)
		, do
			tmpdir <- fromRepo gitAnnexTmpMiscDir
			liftIO $ Lsof.queryDir tmpdir
		)

{- After a Change is committed, queue any necessary transfers or drops
 - of the content of the key.
 -
 - This is not done during the startup scan, because the expensive
 - transfer scan does the same thing then.
 -}
checkChangeContent :: Change -> Assistant ()
checkChangeContent change@(Change { changeInfo = i }) =
	case changeInfoKey i of
		Nothing -> noop
		Just k -> whenM (scanComplete <$> getDaemonStatus) $ do
			present <- liftAnnex $ inAnnex k
			void $ if present
				then queueTransfers "new file created" Next k (Just f) Upload
				else queueTransfers "new or renamed file wanted" Next k (Just f) Download
			handleDrops "file renamed" present k (Just f) Nothing
  where
	f = changeFile change
checkChangeContent _ = noop


code:
{- git-annex assistant config monitor thread
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.ConfigMonitor where

import Assistant.Common
import Assistant.BranchChange
import Assistant.DaemonStatus
import Assistant.Commits
import Utility.ThreadScheduler
import Logs
import Logs.UUID
import Logs.Trust
import Logs.PreferredContent
import Logs.Group
import Logs.NumCopies
import Remote.List (remoteListRefresh)
import qualified Git.LsTree as LsTree
import Git.FilePath
import qualified Annex.Branch

import qualified Data.Set as S

{- This thread detects when configuration changes have been made to the
 - git-annex branch and reloads cached configuration.
 -
 - If the branch is frequently changing, it's checked for configuration
 - changes no more often than once every 60 seconds. On the other hand,
 - if the branch has not changed in a while, configuration changes will
 - be detected immediately.
 -}
configMonitorThread :: NamedThread
configMonitorThread = namedThread "ConfigMonitor" $ loop =<< getConfigs
  where
	loop old = do
		waitBranchChange
		new <- getConfigs
		when (old /= new) $ do
			let changedconfigs = new `S.difference` old
			debug $ "reloading config" :
				map fst (S.toList changedconfigs)
			reloadConfigs new
			{- Record a commit to get this config
			 - change pushed out to remotes. -}
			recordCommit
		liftIO $ threadDelaySeconds (Seconds 60)
		loop new

{- Config files, and their checksums. -}
type Configs = S.Set (FilePath, String)

{- All git-annex's config files, and actions to run when they change. -}
configFilesActions :: [(FilePath, Assistant ())]
configFilesActions =
	[ (uuidLog, void $ liftAnnex uuidMapLoad)
	, (remoteLog, void $ liftAnnex remoteListRefresh)
	, (trustLog, void $ liftAnnex trustMapLoad)
	, (groupLog, void $ liftAnnex groupMapLoad)
	, (numcopiesLog, void $ liftAnnex globalNumCopiesLoad)
	, (scheduleLog, void updateScheduleLog)
	-- Preferred and required content settings depend on most of the
	-- other configs, so will be reloaded whenever any configs change.
	, (preferredContentLog, noop)
	, (requiredContentLog, noop)
	, (groupPreferredContentLog, noop)
	]

reloadConfigs :: Configs -> Assistant ()
reloadConfigs changedconfigs = do
	sequence_ as
	void $ liftAnnex preferredRequiredMapsLoad
	{- Changes to the remote log, or the trust log, can affect the
	 - syncRemotes list. Changes to the uuid log may affect its
	 - display so are also included. -}
	when (any (`elem` fs) [remoteLog, trustLog, uuidLog])
		updateSyncRemotes
  where
	(fs, as) = unzip $ filter (flip S.member changedfiles . fst)
		configFilesActions
	changedfiles = S.map fst changedconfigs

getConfigs :: Assistant Configs
getConfigs = S.fromList . map extract
	<$> liftAnnex (inRepo $ LsTree.lsTreeFiles Annex.Branch.fullname files)
  where
	files = map fst configFilesActions
	extract treeitem = (getTopFilePath $ LsTree.file treeitem, LsTree.sha treeitem)

repo_name: kakafarm/git

license: unknown


code:
{- git-annex assistant sceduled jobs runner
 -
 - Copyright 2013 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

{-# LANGUAGE DeriveDataTypeable #-}

module Assistant.Threads.Cronner (
	cronnerThread
) where

import Assistant.Common
import Assistant.DaemonStatus
import Utility.NotificationBroadcaster
import Annex.UUID
import Config.Files
import Logs.Schedule
import Utility.Scheduled
import Types.ScheduledActivity
import Utility.ThreadScheduler
import Utility.HumanTime
import Utility.Batch
import Assistant.TransferQueue
import Annex.Content
import Logs.Transfer
import Assistant.Types.UrlRenderer
import Assistant.Alert
import Remote
import qualified Types.Remote as Remote
import qualified Git
import qualified Git.Fsck
import Assistant.Fsck
import Assistant.Repair

import Control.Concurrent.Async
import Control.Concurrent.MVar
import Data.Time.LocalTime
import Data.Time.Clock
import qualified Data.Map as M
import qualified Data.Set as S

{- Loads schedules for this repository, and fires off one thread for each
 - scheduled event that runs on this repository. Each thread sleeps until
 - its event is scheduled to run.
 -
 - To handle events that run on remotes, which need to only run when
 - their remote gets connected, threads are also started, and are passed
 - a MVar to wait on, which is stored in the DaemonStatus's
 - connectRemoteNotifiers.
 -
 - In the meantime the main thread waits for any changes to the
 - schedules. When there's a change, compare the old and new list of
 - schedules to find deleted and added ones. Start new threads for added
 - ones, and kill the threads for deleted ones. -}
cronnerThread :: UrlRenderer -> NamedThread
cronnerThread urlrenderer = namedThreadUnchecked "Cronner" $ do
	fsckNudge urlrenderer Nothing
	dstatus <- getDaemonStatus
	h <- liftIO $ newNotificationHandle False (scheduleLogNotifier dstatus)
	go h M.empty M.empty
  where
	go h amap nmap = do
		activities <- liftAnnex $ scheduleGet =<< getUUID

		let addedactivities = activities `S.difference` M.keysSet amap
		let removedactivities = M.keysSet amap `S.difference` activities

		forM_ (S.toList removedactivities) $ \activity ->
			case M.lookup activity amap of
				Just a -> do
					debug ["stopping removed job for", fromScheduledActivity activity, show (asyncThreadId a)]
					liftIO $ cancel a
				Nothing -> noop

		lastruntimes <- liftAnnex getLastRunTimes
		started <- startactivities (S.toList addedactivities) lastruntimes
		let addedamap = M.fromList $ map fst started
		let addednmap = M.fromList $ catMaybes $ map snd started

		let removefiltered = M.filterWithKey (\k _ -> S.member k removedactivities)
		let amap' = M.difference (M.union addedamap amap) (removefiltered amap)
		let nmap' = M.difference (M.union addednmap nmap) (removefiltered nmap)
		modifyDaemonStatus_ $ \s -> s { connectRemoteNotifiers = M.fromListWith (++) (M.elems nmap') }

		liftIO $ waitNotification h
		debug ["reloading changed activities"]
		go h amap' nmap'
  	startactivities as lastruntimes = forM as $ \activity ->
		case connectActivityUUID activity of
			Nothing -> do
				runner <- asIO2 (sleepingActivityThread urlrenderer)
				a <- liftIO $ async $
					runner activity (M.lookup activity lastruntimes)
				return ((activity, a), Nothing)
			Just u -> do
				mvar <- liftIO newEmptyMVar
				runner <- asIO2 (remoteActivityThread urlrenderer mvar)
				a <- liftIO $ async $
					runner activity (M.lookup activity lastruntimes)
				return ((activity, a), Just (activity, (u, [mvar])))

{- Calculate the next time the activity is scheduled to run, then
 - sleep until that time, and run it. Then call setLastRunTime, and
 - loop.
 -}
sleepingActivityThread :: UrlRenderer -> ScheduledActivity -> Maybe LocalTime -> Assistant ()
sleepingActivityThread urlrenderer activity lasttime = go lasttime =<< getnexttime lasttime
  where
  	getnexttime = liftIO . nextTime schedule
  	go _ Nothing = debug ["no scheduled events left for", desc]
	go l (Just (NextTimeExactly t)) = waitrun l t Nothing
	go l (Just (NextTimeWindow windowstart windowend)) =
		waitrun l windowstart (Just windowend)
	desc = fromScheduledActivity activity
	schedule = getSchedule activity
	waitrun l t mmaxt = do
		seconds <- liftIO $ secondsUntilLocalTime t
		when (seconds > Seconds 0) $ do
			debug ["waiting", show seconds, "for next scheduled", desc]
			liftIO $ threadDelaySeconds seconds
		now <- liftIO getCurrentTime
		tz <- liftIO $ getTimeZone now
		let nowt = utcToLocalTime tz now
		if tolate nowt tz
			then do
				debug ["too late to run scheduled", desc]
				go l =<< getnexttime l
			else run nowt
	  where
	  	tolate nowt tz = case mmaxt of
			Just maxt -> nowt > maxt
			-- allow the job to start 10 minutes late
			Nothing ->diffUTCTime
				(localTimeToUTC tz nowt)
				(localTimeToUTC tz t) > 600
	run nowt = do
		runActivity urlrenderer activity nowt
		go (Just nowt) =<< getnexttime (Just nowt)

{- Wait for the remote to become available by waiting on the MVar.
 - Then check if the time is within a time window when activity
 - is scheduled to run, and if so run it.
 - Otherwise, just wait again on the MVar.
 -}
remoteActivityThread :: UrlRenderer -> MVar () -> ScheduledActivity -> Maybe LocalTime -> Assistant ()
remoteActivityThread urlrenderer mvar activity lasttime = do
	liftIO $ takeMVar mvar
	go =<< liftIO (nextTime (getSchedule activity) lasttime)
  where
	go (Just (NextTimeWindow windowstart windowend)) = do
		now <- liftIO getCurrentTime
		tz <- liftIO $ getTimeZone now
		if now >= localTimeToUTC tz windowstart && now <= localTimeToUTC tz windowend
			then do
				let nowt = utcToLocalTime tz now
				runActivity urlrenderer activity nowt
				loop (Just nowt)
			else loop lasttime
	go _ = noop -- running at exact time not handled here
	loop = remoteActivityThread urlrenderer mvar activity

secondsUntilLocalTime :: LocalTime -> IO Seconds
secondsUntilLocalTime t = do
	now <- getCurrentTime
	tz <- getTimeZone now
	let secs = truncate $ diffUTCTime (localTimeToUTC tz t) now
	return $ if secs > 0
		then Seconds secs
		else Seconds 0

runActivity :: UrlRenderer -> ScheduledActivity -> LocalTime -> Assistant ()
runActivity urlrenderer activity nowt = do
	debug ["starting", desc]
	runActivity' urlrenderer activity
	debug ["finished", desc]
	liftAnnex $ setLastRunTime activity nowt
  where
	desc = fromScheduledActivity activity

runActivity' :: UrlRenderer -> ScheduledActivity -> Assistant ()
runActivity' urlrenderer (ScheduledSelfFsck _ d) = do
	program <- liftIO $ readProgramFile
	g <- liftAnnex gitRepo
	fsckresults <- showFscking urlrenderer Nothing $ tryNonAsync $ do
		void $ batchCommand program (Param "fsck" : annexFsckParams d)
		Git.Fsck.findBroken True g
	u <- liftAnnex getUUID
	void $ repairWhenNecessary urlrenderer u Nothing fsckresults
	mapM_ reget =<< liftAnnex (dirKeys gitAnnexBadDir)
  where
	reget k = queueTransfers "fsck found bad file; redownloading" Next k Nothing Download
runActivity' urlrenderer (ScheduledRemoteFsck u s d) = handle =<< liftAnnex (remoteFromUUID u)
  where
	handle Nothing = debug ["skipping remote fsck of uuid without a configured remote", fromUUID u, fromSchedule s]
	handle (Just rmt) = void $ case Remote.remoteFsck rmt of
		Nothing -> go rmt $ do
			program <- readProgramFile
			void $ batchCommand program $
				[ Param "fsck"
				-- avoid downloading files
				, Param "--fast"
				, Param "--from"
				, Param $ Remote.name rmt
				] ++ annexFsckParams d
		Just mkfscker -> do
			{- Note that having mkfsker return an IO action
			 - avoids running a long duration fsck in the
			 - Annex monad. -}
			go rmt =<< liftAnnex (mkfscker (annexFsckParams d))
	go rmt annexfscker = do
		fsckresults <- showFscking urlrenderer (Just rmt) $ tryNonAsync $ do
			void annexfscker
			let r = Remote.repo rmt
			if Git.repoIsLocal r && not (Git.repoIsLocalUnknown r)
				then Just <$> Git.Fsck.findBroken True r
				else pure Nothing
		maybe noop (void . repairWhenNecessary urlrenderer u (Just rmt)) fsckresults

annexFsckParams :: Duration -> [CommandParam]
annexFsckParams d =
	[ Param "--incremental-schedule=1d"
	, Param $ "--time-limit=" ++ fromDuration d
	]


code:
{- git-annex assistant daemon status thread
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.DaemonStatus where

import Assistant.Common
import Assistant.DaemonStatus
import Utility.ThreadScheduler
import Utility.NotificationBroadcaster

{- This writes the daemon status to disk, when it changes, but no more
 - frequently than once every ten minutes.
 -}
daemonStatusThread :: NamedThread
daemonStatusThread = namedThread "DaemonStatus" $ do
	notifier <- liftIO . newNotificationHandle False
		=<< changeNotifier <$> getDaemonStatus
	checkpoint
	runEvery (Seconds tenMinutes) <~> do
		liftIO $ waitNotification notifier
		checkpoint
  where
	checkpoint = do
		file <- liftAnnex $ fromRepo gitAnnexDaemonStatusFile
		liftIO . writeDaemonStatusFile file =<< getDaemonStatus


code:
{- git-annex assistant Amazon Glacier retrieval
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

{-# LANGUAGE CPP #-}
{-# LANGUAGE OverloadedStrings #-}

module Assistant.Threads.Glacier where

import Assistant.Common
import Utility.ThreadScheduler
import qualified Types.Remote as Remote
import qualified Remote.Glacier as Glacier
import Logs.Transfer
import Assistant.DaemonStatus
import Assistant.TransferQueue

import qualified Data.Set as S

{- Wakes up every half hour and checks if any glacier remotes have failed
 - downloads. If so, runs glacier-cli to check if the files are now
 - available, and queues the downloads. -}
glacierThread :: NamedThread
glacierThread = namedThread "Glacier" $ runEvery (Seconds 3600) <~> go
  where
	isglacier r = Remote.remotetype r == Glacier.remote
	go = do
		rs <- filter isglacier . syncDataRemotes <$> getDaemonStatus
		forM_ rs $ \r ->
			check r =<< liftAnnex (getFailedTransfers $ Remote.uuid r)
	check _ [] = noop
	check r l = do
		let keys = map getkey l
		(availkeys, failedkeys) <- liftAnnex $ Glacier.jobList r keys
		let s = S.fromList (failedkeys ++ availkeys)
		let l' = filter (\p -> S.member (getkey p) s) l
		forM_ l' $ \(t, info) -> do
			liftAnnex $ removeFailedTransfer t
			queueTransferWhenSmall "object available from glacier" (associatedFile info) t r
	getkey = transferKey . fst


code:
{- git-annex assistant git merge thread
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.Merger where

import Assistant.Common
import Assistant.TransferQueue
import Assistant.BranchChange
import Assistant.DaemonStatus
import Assistant.ScanRemotes
import Utility.DirWatcher
import Utility.DirWatcher.Types
import qualified Annex.Branch
import qualified Git
import qualified Git.Branch
import Annex.AutoMerge
import Annex.TaggedPush
import Remote (remoteFromUUID)

import qualified Data.Set as S
import qualified Data.Text as T

{- This thread watches for changes to .git/refs/, and handles incoming
 - pushes. -}
mergeThread :: NamedThread
mergeThread = namedThread "Merger" $ do
	g <- liftAnnex gitRepo
	let dir = Git.localGitDir g </> "refs"
	liftIO $ createDirectoryIfMissing True dir
	let hook a = Just <$> asIO2 (runHandler a)
	changehook <- hook onChange
	errhook <- hook onErr
	let hooks = mkWatchHooks
		{ addHook = changehook
		, modifyHook = changehook
		, errHook = errhook
		}
	void $ liftIO $ watchDir dir (const False) True hooks id
	debug ["watching", dir]

type Handler = FilePath -> Assistant ()

{- Runs an action handler.
 -
 - Exceptions are ignored, otherwise a whole thread could be crashed.
 -}
runHandler :: Handler -> FilePath -> Maybe FileStatus -> Assistant ()
runHandler handler file _filestatus =
	either (liftIO . print) (const noop) =<< tryIO <~> handler file

{- Called when there's an error with inotify. -}
onErr :: Handler
onErr = error

{- Called when a new branch ref is written, or a branch ref is modified.
 -
 - At startup, synthetic add events fire, causing this to run, but that's
 - ok; it ensures that any changes pushed since the last time the assistant
 - ran are merged in.
 -}
onChange :: Handler
onChange file
	| ".lock" `isSuffixOf` file = noop
	| isAnnexBranch file = do
		branchChanged
		diverged <- liftAnnex Annex.Branch.forceUpdate
		when diverged $
			unlessM handleDesynced $
				queueDeferredDownloads "retrying deferred download" Later
	| "/synced/" `isInfixOf` file =
		mergecurrent =<< liftAnnex (inRepo Git.Branch.current)
	| otherwise = noop
  where
	changedbranch = fileToBranch file

	mergecurrent (Just current)
		| equivBranches changedbranch current =
			whenM (liftAnnex $ inRepo $ Git.Branch.changed current changedbranch) $ do
				debug
					[ "merging", Git.fromRef changedbranch
					, "into", Git.fromRef current
					]
				void $ liftAnnex $ autoMergeFrom changedbranch (Just current) Git.Branch.AutomaticCommit
	mergecurrent _ = noop

	handleDesynced = case fromTaggedBranch changedbranch of
		Nothing -> return False
		Just (u, info) -> do
			mr <- liftAnnex $ remoteFromUUID u
			case mr of
				Nothing -> return False
				Just r -> do
					s <- desynced <$> getDaemonStatus
					if S.member u s || Just (T.unpack $ getXMPPClientID r) == info
						then do
							modifyDaemonStatus_ $ \st -> st
								{ desynced = S.delete u s }
							addScanRemotes True [r]
							return True
						else return False

equivBranches :: Git.Ref -> Git.Ref -> Bool
equivBranches x y = base x == base y
  where
	base = takeFileName . Git.fromRef

isAnnexBranch :: FilePath -> Bool
isAnnexBranch f = n `isSuffixOf` f
  where
	n = '/' : Git.fromRef Annex.Branch.name

fileToBranch :: FilePath -> Git.Ref
fileToBranch f = Git.Ref $ "refs" </> base
  where
	base = Prelude.last $ split "/refs/" f


code:
{- git-annex assistant mount watcher, using either dbus or mtab polling
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

{-# LANGUAGE CPP #-}
{-# LANGUAGE OverloadedStrings #-}

module Assistant.Threads.MountWatcher where

import Assistant.Common
import Assistant.DaemonStatus
import Assistant.Sync
import qualified Annex
import qualified Git
import Utility.ThreadScheduler
import Utility.Mounts
import Remote.List
import qualified Types.Remote as Remote
import Assistant.Types.UrlRenderer
import Assistant.Fsck

import qualified Data.Set as S

#if WITH_DBUS
import Utility.DBus
import DBus.Client
import DBus
import Data.Word (Word32)
import Control.Concurrent
import qualified Control.Exception as E
#else
#warning Building without dbus support; will use mtab polling
#endif

mountWatcherThread :: UrlRenderer -> NamedThread
mountWatcherThread urlrenderer = namedThread "MountWatcher" $
#if WITH_DBUS
	dbusThread urlrenderer
#else
	pollingThread urlrenderer
#endif

#if WITH_DBUS

dbusThread :: UrlRenderer -> Assistant ()
dbusThread urlrenderer = do
	runclient <- asIO1 go
	r <- liftIO $ E.try $ runClient getSessionAddress runclient
	either onerr (const noop) r
  where
	go client = ifM (checkMountMonitor client)
		( do
			{- Store the current mount points in an MVar, to be
			 - compared later. We could in theory work out the
			 - mount point from the dbus message, but this is
			 - easier. -}
			mvar <- liftIO $ newMVar =<< currentMountPoints
			handleevent <- asIO1 $ \_event -> do
				nowmounted <- liftIO $ currentMountPoints
				wasmounted <- liftIO $ swapMVar mvar nowmounted
				handleMounts urlrenderer wasmounted nowmounted
			liftIO $ forM_ mountChanged $ \matcher ->
#if MIN_VERSION_dbus(0,10,7)
				void $ addMatch client matcher handleevent
#else
				listen client matcher handleevent
#endif
		, do
			liftAnnex $
				warning "No known volume monitor available through dbus; falling back to mtab polling"
			pollingThread urlrenderer
		)
	onerr :: E.SomeException -> Assistant ()
	onerr e = do
		{- If the session dbus fails, the user probably
		 - logged out of their desktop. Even if they log
		 - back in, we won't have access to the dbus
		 - session key, so polling is the best that can be
		 - done in this situation. -}
		liftAnnex $
			warning $ "dbus failed; falling back to mtab polling (" ++ show e ++ ")"
		pollingThread urlrenderer

{- Examine the list of services connected to dbus, to see if there
 - are any we can use to monitor mounts. If not, will attempt to start one. -}
checkMountMonitor :: Client -> Assistant Bool
checkMountMonitor client = do
	running <- filter (`elem` usableservices)
		<$> liftIO (listServiceNames client)
	case running of
		[] -> startOneService client startableservices
		(service:_) -> do
			debug [ "Using running DBUS service"
				, service
				, "to monitor mount events."
				]
			return True
  where
	startableservices = [gvfs, gvfsgdu]
	usableservices = startableservices ++ [kde]
	gvfs = "org.gtk.Private.UDisks2VolumeMonitor"
	gvfsgdu = "org.gtk.Private.GduVolumeMonitor"
	kde = "org.kde.DeviceNotifications"

startOneService :: Client -> [ServiceName] -> Assistant Bool
startOneService _ [] = return False
startOneService client (x:xs) = do
	_ <- liftIO $ tryNonAsync $ callDBus client "StartServiceByName"
		[toVariant x, toVariant (0 :: Word32)]
	ifM (liftIO $ elem x <$> listServiceNames client)
		( do
			debug
				[ "Started DBUS service", x
				, "to monitor mount events."
				]
			return True
		, startOneService client xs
		)

{- Filter matching events recieved when drives are mounted and unmounted. -}
mountChanged :: [MatchRule]
mountChanged = [gvfs True, gvfs False, kde, kdefallback]
  where
	{- gvfs reliably generates this event whenever a
	 - drive is mounted/unmounted, whether automatically, or manually -}
	gvfs mount = matchAny
		{ matchInterface = Just "org.gtk.Private.RemoteVolumeMonitor"
		, matchMember = Just $ if mount then "MountAdded" else "MountRemoved"
		}
	{- This event fires when KDE prompts the user what to do with a drive,
	 - but maybe not at other times. And it's not received -}
	kde = matchAny
		{ matchInterface = Just "org.kde.Solid.Device"
		, matchMember = Just "setupDone"
		}
	{- This event may not be closely related to mounting a drive, but it's
	 - observed reliably when a drive gets mounted or unmounted. -}
	kdefallback = matchAny
		{ matchInterface = Just "org.kde.KDirNotify"
		, matchMember = Just "enteredDirectory"
		}

#endif

pollingThread :: UrlRenderer -> Assistant ()
pollingThread urlrenderer = go =<< liftIO currentMountPoints
  where
	go wasmounted = do
		liftIO $ threadDelaySeconds (Seconds 10)
		nowmounted <- liftIO currentMountPoints
		handleMounts urlrenderer wasmounted nowmounted
		go nowmounted

handleMounts :: UrlRenderer -> MountPoints -> MountPoints -> Assistant ()
handleMounts urlrenderer wasmounted nowmounted =
	mapM_ (handleMount urlrenderer . mnt_dir) $
		S.toList $ newMountPoints wasmounted nowmounted

handleMount :: UrlRenderer -> FilePath -> Assistant ()
handleMount urlrenderer dir = do
	debug ["detected mount of", dir]
	rs <- filter (Git.repoIsLocal . Remote.repo) <$> remotesUnder dir
	mapM_ (fsckNudge urlrenderer . Just) rs
	reconnectRemotes True rs

{- Finds remotes located underneath the mount point.
 -
 - Updates state to include the remotes.
 -
 - The config of git remotes is re-read, as it may not have been available
 - at startup time, or may have changed (it could even be a different
 - repository at the same remote location..)
 -}
remotesUnder :: FilePath -> Assistant [Remote]
remotesUnder dir = do
	repotop <- liftAnnex $ fromRepo Git.repoPath
	rs <- liftAnnex remoteList
	pairs <- liftAnnex $ mapM (checkremote repotop) rs
	let (waschanged, rs') = unzip pairs
	when (or waschanged) $ do
		liftAnnex $ Annex.changeState $ \s -> s { Annex.remotes = catMaybes rs' }
		updateSyncRemotes
	return $ mapMaybe snd $ filter fst pairs
  where
	checkremote repotop r = case Remote.localpath r of
		Just p | dirContains dir (absPathFrom repotop p) ->
			(,) <$> pure True <*> updateRemote r
		_ -> return (False, Just r)

type MountPoints = S.Set Mntent

currentMountPoints :: IO MountPoints
currentMountPoints = S.fromList <$> getMounts

newMountPoints :: MountPoints -> MountPoints -> MountPoints
newMountPoints old new = S.difference new old


code:
{- git-annex assistant network connection watcher, using dbus
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

{-# LANGUAGE CPP #-}
{-# LANGUAGE OverloadedStrings #-}

module Assistant.Threads.NetWatcher where

import Assistant.Common
import Assistant.Sync
import Utility.ThreadScheduler
import qualified Types.Remote as Remote
import Assistant.DaemonStatus
import Assistant.RemoteControl
import Utility.NotificationBroadcaster

#if WITH_DBUS
import Utility.DBus
import DBus.Client
import DBus
import Assistant.NetMessager
#else
#ifdef linux_HOST_OS
#warning Building without dbus support; will poll for network connection changes
#endif
#endif

netWatcherThread :: NamedThread
#if WITH_DBUS
netWatcherThread = thread dbusThread
#else
netWatcherThread = thread noop
#endif
  where
	thread = namedThread "NetWatcher"

{- This is a fallback for when dbus cannot be used to detect
 - network connection changes, but it also ensures that
 - any networked remotes that may have not been routable for a
 - while (despite the local network staying up), are synced with
 - periodically.
 -
 - Note that it does not call notifyNetMessagerRestart, or
 - signal the RemoteControl, because it doesn't know that the
 - network has changed.
 -}
netWatcherFallbackThread :: NamedThread
netWatcherFallbackThread = namedThread "NetWatcherFallback" $
	runEvery (Seconds 3600) <~> handleConnection

#if WITH_DBUS

dbusThread :: Assistant ()
dbusThread = do
	handleerr <- asIO2 onerr
	runclient <- asIO1 go
	liftIO $ persistentClient getSystemAddress () handleerr runclient
  where
	go client = ifM (checkNetMonitor client)
		( do
			callback <- asIO1 connchange
			liftIO $ do
				listenNMConnections client callback
				listenWicdConnections client callback
		, do
			liftAnnex $
				warning "No known network monitor available through dbus; falling back to polling"
		)
	connchange False = do
		debug ["detected network disconnection"]
		sendRemoteControl LOSTNET
	connchange True = do
		debug ["detected network connection"]
		notifyNetMessagerRestart
		handleConnection
		sendRemoteControl RESUME
	onerr e _ = do
		liftAnnex $
			warning $ "lost dbus connection; falling back to polling (" ++ show e ++ ")"
		{- Wait, in hope that dbus will come back -}
		liftIO $ threadDelaySeconds (Seconds 60)

{- Examine the list of services connected to dbus, to see if there
 - are any we can use to monitor network connections. -}
checkNetMonitor :: Client -> Assistant Bool
checkNetMonitor client = do
	running <- liftIO $ filter (`elem` [networkmanager, wicd])
		<$> listServiceNames client
	case running of
		[] -> return False
		(service:_) -> do
			debug [ "Using running DBUS service"
				, service
				, "to monitor network connection events."
				]
			return True
  where
	networkmanager = "org.freedesktop.NetworkManager"
	wicd = "org.wicd.daemon"

{- Listens for NetworkManager connections and diconnections.
 -
 - Connection example (once fully connected):
 - [Variant {"ActivatingConnection": Variant (ObjectPath "/"), "PrimaryConnection": Variant (ObjectPath "/org/freedesktop/NetworkManager/ActiveConnection/34"), "State": Variant 70}]
 -
 - Disconnection example:
 - [Variant {"ActiveConnections": Variant []}]
 -}
listenNMConnections :: Client -> (Bool -> IO ()) -> IO ()
listenNMConnections client setconnected =
#if MIN_VERSION_dbus(0,10,7)
	void $ addMatch client matcher
#else
	listen client matcher
#endif
		$ \event -> mapM_ handle
			(map dictionaryItems $ mapMaybe fromVariant $ signalBody event)
  where
	matcher = matchAny
		{ matchInterface = Just "org.freedesktop.NetworkManager"
		, matchMember = Just "PropertiesChanged"
		}
	nm_active_connections_key = toVariant ("ActiveConnections" :: String)
	nm_activatingconnection_key = toVariant ("ActivatingConnection" :: String)
	noconnections = Just $ toVariant $ toVariant ([] :: [ObjectPath])
	rootconnection = Just $ toVariant $ toVariant $ objectPath_ "/"
	handle m
		| lookup nm_active_connections_key m == noconnections =
			setconnected False
		| lookup nm_activatingconnection_key m == rootconnection =
			setconnected True
		| otherwise = noop

{- Listens for Wicd connections and disconnections.
 -
 - Connection example:
 -   ConnectResultsSent:
 -     Variant "success"
 -
 - Diconnection example:
 -   StatusChanged
 -     [Variant 0, Variant [Varient ""]]
 -}
listenWicdConnections :: Client -> (Bool -> IO ()) -> IO ()
listenWicdConnections client setconnected = do
	match connmatcher $ \event ->
		when (any (== wicd_success) (signalBody event)) $
			setconnected True
	match statusmatcher $ \event -> handle (signalBody event)
  where
	connmatcher = matchAny
		{ matchInterface = Just "org.wicd.daemon"
		, matchMember = Just "ConnectResultsSent"
		}
	statusmatcher = matchAny
		{ matchInterface = Just "org.wicd.daemon"
		, matchMember = Just "StatusChanged"
		}
	wicd_success = toVariant ("success" :: String)
	wicd_disconnected = toVariant [toVariant ("" :: String)]
	handle status
		| any (== wicd_disconnected) status = setconnected False
		| otherwise = noop
	match matcher a =
#if MIN_VERSION_dbus(0,10,7)
		void $ addMatch client matcher a
#else
		listen client matcher a
#endif
#endif

handleConnection :: Assistant ()
handleConnection = do
	liftIO . sendNotification . networkConnectedNotifier =<< getDaemonStatus
	reconnectRemotes True =<< networkRemotes

{- Network remotes to sync with. -}
networkRemotes :: Assistant [Remote]
networkRemotes = filter (isNothing . Remote.localpath) . syncRemotes
	<$> getDaemonStatus


code:
{- git-annex assistant thread to listen for incoming pairing traffic
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.PairListener where

import Assistant.Common
import Assistant.Pairing
import Assistant.Pairing.Network
import Assistant.Pairing.MakeRemote
import Assistant.WebApp (UrlRenderer)
import Assistant.WebApp.Types
import Assistant.Alert
import Assistant.DaemonStatus
import Utility.ThreadScheduler
import Utility.Format
import Git

import Network.Multicast
import Network.Socket
import qualified Data.Text as T
import Data.Char

pairListenerThread :: UrlRenderer -> NamedThread
pairListenerThread urlrenderer = namedThread "PairListener" $ do
	listener <- asIO1 $ go [] []
	liftIO $ withSocketsDo $
		runEvery (Seconds 60) $ void $ tryIO $
			listener =<< getsock
  where
	{- Note this can crash if there's no network interface,
	 - or only one like lo that doesn't support multicast. -}
	getsock = multicastReceiver (multicastAddress $ IPv4Addr undefined) pairingPort

	go reqs cache sock = liftIO (getmsg sock []) >>= \msg -> case readish msg of
		Nothing -> go reqs cache sock
		Just m -> do
			debug ["received", show msg]
			sane <- checkSane msg
			(pip, verified) <- verificationCheck m
				=<< (pairingInProgress <$> getDaemonStatus)
			let wrongstage = maybe False (\p -> pairMsgStage m <= inProgressPairStage p) pip
			let fromus = maybe False (\p -> remoteSshPubKey (pairMsgData m) == remoteSshPubKey (inProgressPairData p)) pip
			case (wrongstage, fromus, sane, pairMsgStage m) of
				(_, True, _, _) -> do
					debug ["ignoring message that looped back"]
					go reqs cache sock
				(_, _, False, _) -> go reqs cache sock
				-- PairReq starts a pairing process, so a
				-- new one is always heeded, even if
				-- some other pairing is in process.
				(_, _, _, PairReq) -> if m `elem` reqs
					then go reqs (invalidateCache m cache) sock
					else do
						pairReqReceived verified urlrenderer m
						go (m:take 10 reqs) (invalidateCache m cache) sock
				(True, _, _, _) -> do
					debug
						["ignoring out of order message"
						, show (pairMsgStage m)
						, "expected"
						, show (succ . inProgressPairStage <$> pip)
						]
					go reqs cache sock
				(_, _, _, PairAck) -> do
					cache' <- pairAckReceived verified pip m cache
					go reqs cache' sock
				(_,_ , _, PairDone) -> do
					pairDoneReceived verified pip m
					go reqs cache sock

	{- As well as verifying the message using the shared secret,
	 - check its UUID against the UUID we have stored. If
	 - they're the same, someone is sending bogus messages,
	 - which could be an attempt to brute force the shared secret. -}
	verificationCheck _ Nothing = return (Nothing, False)
	verificationCheck m (Just pip)
		| not verified && sameuuid = do
			liftAnnex $ warning
				"detected possible pairing brute force attempt; disabled pairing"
			stopSending pip
			return (Nothing, False)
		|otherwise = return (Just pip, verified && sameuuid)
	  where
		verified = verifiedPairMsg m pip
		sameuuid = pairUUID (inProgressPairData pip) == pairUUID (pairMsgData m)

	checkSane msg
		{- Control characters could be used in a
		 - console poisoning attack. -}
		| any isControl (filter (/= '\n') (decode_c msg)) = do
			liftAnnex $ warning
				"illegal control characters in pairing message; ignoring"
			return False
		| otherwise = return True

	{- PairReqs invalidate the cache of recently finished pairings.
	 - This is so that, if a new pairing is started with the
	 - same secret used before, a bogus PairDone is not sent. -}
	invalidateCache msg = filter (not . verifiedPairMsg msg)

	getmsg sock c = do
		(msg, n, _) <- recvFrom sock chunksz
		if n < chunksz
			then return $ c ++ msg
			else getmsg sock $ c ++ msg
	  where
		chunksz = 1024

{- Show an alert when a PairReq is seen. -}
pairReqReceived :: Bool -> UrlRenderer -> PairMsg -> Assistant ()
pairReqReceived True _ _ = noop -- ignore our own PairReq
pairReqReceived False urlrenderer msg = do
	button <- mkAlertButton True (T.pack "Respond") urlrenderer (FinishLocalPairR msg)
	void $ addAlert $ pairRequestReceivedAlert repo button
  where
	repo = pairRepo msg

{- When a verified PairAck is seen, a host is ready to pair with us, and has
 - already configured our ssh key. Stop sending PairReqs, finish the pairing,
 - and send a single PairDone. -}
pairAckReceived :: Bool -> Maybe PairingInProgress -> PairMsg -> [PairingInProgress] -> Assistant [PairingInProgress]
pairAckReceived True (Just pip) msg cache = do
	stopSending pip
	repodir <- repoPath <$> liftAnnex gitRepo
	liftIO $ setupAuthorizedKeys msg repodir
	finishedLocalPairing msg (inProgressSshKeyPair pip)
	startSending pip PairDone $ multicastPairMsg
		(Just 1) (inProgressSecret pip) (inProgressPairData pip)
	return $ pip : take 10 cache
{- A stale PairAck might also be seen, after we've finished pairing.
 - Perhaps our PairDone was not received. To handle this, we keep
 - a cache of recently finished pairings, and re-send PairDone in
 - response to stale PairAcks for them. -}
pairAckReceived _ _ msg cache = do
	let pips = filter (verifiedPairMsg msg) cache
	unless (null pips) $
		forM_ pips $ \pip ->
			startSending pip PairDone $ multicastPairMsg
				(Just 1) (inProgressSecret pip) (inProgressPairData pip)
	return cache

{- If we get a verified PairDone, the host has accepted our PairAck, and
 - has paired with us. Stop sending PairAcks, and finish pairing with them.
 -
 - TODO: Should third-party hosts remove their pair request alert when they
 - see a PairDone?
 - Complication: The user could have already clicked on the alert and be
 - entering the secret. Would be better to start a fresh pair request in this
 - situation.
 -}
pairDoneReceived :: Bool -> Maybe PairingInProgress -> PairMsg -> Assistant ()
pairDoneReceived False _ _ = noop -- not verified
pairDoneReceived True Nothing _ = noop -- not in progress
pairDoneReceived True (Just pip) msg = do
	stopSending pip
	finishedLocalPairing msg (inProgressSshKeyPair pip)


code:
{- git-annex assistant thread to handle fixing problems with repositories
 -
 - Copyright 2013 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.ProblemFixer (
	problemFixerThread
) where

import Assistant.Common
import Assistant.Types.RepoProblem
import Assistant.RepoProblem
import Assistant.Types.UrlRenderer
import Assistant.Alert
import Remote
import qualified Types.Remote as Remote
import qualified Git.Fsck
import Assistant.Repair
import qualified Git
import Annex.UUID
import Utility.ThreadScheduler

{- Waits for problems with a repo, and tries to fsck the repo and repair
 - the problem. -}
problemFixerThread :: UrlRenderer -> NamedThread
problemFixerThread urlrenderer = namedThread "ProblemFixer" $
	go =<< getRepoProblems
  where
	go problems = do
		mapM_ (handleProblem urlrenderer) problems
		liftIO $ threadDelaySeconds (Seconds 60)
		-- Problems may have been re-reported while they were being
		-- fixed, so ignore those. If a new unique problem happened
		-- 60 seconds after the last was fixed, we're unlikely
		-- to do much good anyway.
		go =<< filter (\p -> not (any (sameRepoProblem p) problems))
			<$> getRepoProblems

handleProblem :: UrlRenderer -> RepoProblem -> Assistant ()
handleProblem urlrenderer repoproblem = do
	fixed <- ifM ((==) (problemUUID repoproblem) <$> liftAnnex getUUID)
		( handleLocalRepoProblem urlrenderer
		, maybe (return False) (handleRemoteProblem urlrenderer)
			=<< liftAnnex (remoteFromUUID $ problemUUID repoproblem)
		)
	when fixed $
		liftIO $ afterFix repoproblem

handleRemoteProblem :: UrlRenderer -> Remote -> Assistant Bool
handleRemoteProblem urlrenderer rmt
	| Git.repoIsLocal r && not (Git.repoIsLocalUnknown r) =
		ifM (liftIO $ checkAvailable True rmt)
			( do
				fixedlocks <- repairStaleGitLocks r
				fsckresults <- showFscking urlrenderer (Just rmt) $ tryNonAsync $
					Git.Fsck.findBroken True r
				repaired <- repairWhenNecessary urlrenderer (Remote.uuid rmt) (Just rmt) fsckresults
				return $ fixedlocks || repaired
			, return False
			)
	| otherwise = return False
  where
	r = Remote.repo rmt

{- This is not yet used, and should probably do a fsck. -}
handleLocalRepoProblem :: UrlRenderer -> Assistant Bool
handleLocalRepoProblem _urlrenderer = do
	repairStaleGitLocks =<< liftAnnex gitRepo


code:
{- git-annex assistant git pushing thread
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.Pusher where

import Assistant.Common
import Assistant.Commits
import Assistant.Pushes
import Assistant.DaemonStatus
import Assistant.Sync
import Utility.ThreadScheduler
import qualified Remote
import qualified Types.Remote as Remote

{- This thread retries pushes that failed before. -}
pushRetryThread :: NamedThread
pushRetryThread = namedThread "PushRetrier" $ runEvery (Seconds halfhour) <~> do
	-- We already waited half an hour, now wait until there are failed
	-- pushes to retry.
	topush <- getFailedPushesBefore (fromIntegral halfhour)
	unless (null topush) $ do
		debug ["retrying", show (length topush), "failed pushes"]
		void $ pushToRemotes True topush
  where
	halfhour = 1800

{- This thread pushes git commits out to remotes soon after they are made. -}
pushThread :: NamedThread
pushThread = namedThread "Pusher" $ runEvery (Seconds 2) <~> do
	-- We already waited two seconds as a simple rate limiter.
	-- Next, wait until at least one commit has been made
	void getCommits
	-- Now see if now's a good time to push.
	void $ pushToRemotes True =<< pushTargets

{- We want to avoid pushing to remotes that are marked readonly.
 -
 - Also, avoid pushing to local remotes we can easily tell are not available,
 - to avoid ugly messages when a removable drive is not attached.
 -}
pushTargets :: Assistant [Remote]
pushTargets = liftIO . filterM (Remote.checkAvailable True)
	=<< candidates <$> getDaemonStatus
  where
	candidates = filter (not . Remote.readonly) . syncGitRemotes


code:
{- git-annex assistant communication with remotedaemon
 -
 - Copyright 2014 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.RemoteControl where

import Assistant.Common
import RemoteDaemon.Types
import Config.Files
import Utility.Batch
import Utility.SimpleProtocol
import Assistant.Alert
import Assistant.Alert.Utility
import Assistant.DaemonStatus
import qualified Git
import qualified Git.Types as Git
import qualified Remote
import qualified Types.Remote as Remote

import Control.Concurrent
import Control.Concurrent.Async
import Network.URI
import qualified Data.Map as M
import qualified Data.Set as S

remoteControlThread :: NamedThread
remoteControlThread = namedThread "RemoteControl" $ do
	program <- liftIO readProgramFile
	(cmd, params) <- liftIO $ toBatchCommand
		(program, [Param "remotedaemon"])
	let p = proc cmd (toCommand params)
	(Just toh, Just fromh, _, pid) <- liftIO $ createProcess p
		{ std_in = CreatePipe
		, std_out = CreatePipe
		}

	urimap <- liftIO . newMVar =<< liftAnnex getURIMap

	controller <- asIO $ remoteControllerThread toh
	responder <- asIO $ remoteResponderThread fromh urimap

	-- run controller and responder until the remotedaemon dies
	liftIO $ void $ tryNonAsync $ controller `concurrently` responder
	debug ["remotedaemon exited"]
	liftIO $ forceSuccessProcess p pid

-- feed from the remoteControl channel into the remotedaemon
remoteControllerThread :: Handle -> Assistant ()
remoteControllerThread toh = do
	clicker <- getAssistant remoteControl
	forever $ do
		msg <- liftIO $ readChan clicker
		debug [show msg]
		liftIO $ do
			hPutStrLn toh $ unwords $ formatMessage msg
			hFlush toh

-- read status messages emitted by the remotedaemon and handle them
remoteResponderThread :: Handle -> MVar (M.Map URI Remote) -> Assistant ()
remoteResponderThread fromh urimap = go M.empty
  where
	go syncalerts = do
		l <- liftIO $ hGetLine fromh
		debug [l]
		case parseMessage l of
			Just (CONNECTED uri) -> changeconnected S.insert uri
			Just (DISCONNECTED uri) -> changeconnected S.delete uri
			Just (SYNCING uri) -> withr uri $ \r ->
				if M.member (Remote.uuid r) syncalerts
					then go syncalerts
					else do
						i <- addAlert $ syncAlert [r]
						go (M.insert (Remote.uuid r) i syncalerts)
			Just (DONESYNCING uri status) -> withr uri $ \r ->
				case M.lookup (Remote.uuid r) syncalerts of
					Nothing -> cont
					Just i -> do
						let (succeeded, failed) = if status
							then ([r], [])
							else ([], [r])
						updateAlertMap $ mergeAlert i $
							syncResultAlert succeeded failed
						go (M.delete (Remote.uuid r) syncalerts)
			Just (WARNING (RemoteURI uri) msg) -> do
				void $ addAlert $
					warningAlert ("RemoteControl "++ show uri) msg
				cont
			Nothing -> do
				debug ["protocol error from remotedaemon: ", l]
				cont
	  where
		cont = go syncalerts
		withr uri = withRemote uri urimap cont
		changeconnected sm uri = withr uri $ \r -> do
			changeCurrentlyConnected $ sm $ Remote.uuid r
			cont

getURIMap :: Annex (M.Map URI Remote)
getURIMap = Remote.remoteMap' id (mkk . Git.location . Remote.repo)
  where
	mkk (Git.Url u) = Just u
	mkk _ = Nothing

withRemote
	:: RemoteURI
	-> MVar (M.Map URI Remote)
	-> Assistant a
	-> (Remote -> Assistant a)
	-> Assistant a
withRemote (RemoteURI uri) remotemap noremote a = do
	m <- liftIO $ readMVar remotemap
	case M.lookup uri m of
		Just r -> a r
		Nothing -> do
			{- Reload map, in case a new remote has been added. -}
			m' <- liftAnnex getURIMap
			void $ liftIO $ swapMVar remotemap $ m'
			maybe noremote a (M.lookup uri m')


code:
{- git-annex assistant sanity checker
 -
 - Copyright 2012, 2013 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

{-# LANGUAGE CPP #-}

module Assistant.Threads.SanityChecker (
	sanityCheckerStartupThread,
	sanityCheckerDailyThread,
	sanityCheckerHourlyThread
) where

import Assistant.Common
import Assistant.DaemonStatus
import Assistant.Alert
import Assistant.Repair
import Assistant.Drop
import Assistant.Ssh
import Assistant.TransferQueue
import Assistant.Types.UrlRenderer
import qualified Annex.Branch
import qualified Git.LsFiles
import qualified Git.Command.Batch
import qualified Git.Config
import Utility.ThreadScheduler
import qualified Assistant.Threads.Watcher as Watcher
import Utility.Batch
import Utility.NotificationBroadcaster
import Config
import Utility.HumanTime
import Utility.Tense
import Git.Repair
import Git.Index
import Assistant.Unused
import Logs.Unused
import Logs.Transfer
import Config.Files
import Utility.DiskFree
import qualified Annex
import Annex.Exception
#ifdef WITH_WEBAPP
import Assistant.WebApp.Types
#endif
#ifndef mingw32_HOST_OS
import Utility.LogFile
#endif
import Types.Key (keyBackendName)

import Data.Time.Clock.POSIX
import qualified Data.Text as T

{- This thread runs once at startup, and most other threads wait for it
 - to finish. (However, the webapp thread does not, to prevent the UI
 - being nonresponsive.) -}
sanityCheckerStartupThread :: Maybe Duration -> NamedThread
sanityCheckerStartupThread startupdelay = namedThreadUnchecked "SanityCheckerStartup" $ do
	{- Stale git locks can prevent commits from happening, etc. -}
	void $ repairStaleGitLocks =<< liftAnnex gitRepo

	{- A corrupt index file can prevent the assistant from working at
	 - all, so detect and repair. -}
	ifM (not <$> liftAnnex (inRepo checkIndexFast))
		( do
			notice ["corrupt index file found at startup; removing and restaging"]
			liftAnnex $ inRepo $ nukeFile . indexFile
			{- Normally the startup scan avoids re-staging files,
			 - but with the index deleted, everything needs to be
			 - restaged. -}
			modifyDaemonStatus_ $ \s -> s { forceRestage = True }
		, whenM (liftAnnex $ inRepo missingIndex) $ do
			debug ["no index file; restaging"]
			modifyDaemonStatus_ $ \s -> s { forceRestage = True }
		)
	{- If the git-annex index file is corrupt, it's ok to remove it;
	 - the data from the git-annex branch will be used, and the index
	 - will be automatically regenerated. -}
	unlessM (liftAnnex $ Annex.Branch.withIndex $ inRepo $ Git.Repair.checkIndexFast) $ do
		notice ["corrupt annex/index file found at startup; removing"]
		liftAnnex $ liftIO . nukeFile =<< fromRepo gitAnnexIndex

	{- Fix up ssh remotes set up by past versions of the assistant. -}
	liftIO $ fixUpSshRemotes

	{- Clean up old temp files. -}
	void $ liftAnnex $ tryAnnex $ do
		cleanOldTmpMisc
		cleanReallyOldTmp

	{- If there's a startup delay, it's done here. -}
	liftIO $ maybe noop (threadDelaySeconds . Seconds . fromIntegral . durationSeconds) startupdelay

	{- Notify other threads that the startup sanity check is done. -}
	status <- getDaemonStatus
	liftIO $ sendNotification $ startupSanityCheckNotifier status

{- This thread wakes up hourly for inxepensive frequent sanity checks. -}
sanityCheckerHourlyThread :: NamedThread
sanityCheckerHourlyThread = namedThread "SanityCheckerHourly" $ forever $ do
	liftIO $ threadDelaySeconds $ Seconds oneHour
	hourlyCheck

{- This thread wakes up daily to make sure the tree is in good shape. -}
sanityCheckerDailyThread :: UrlRenderer -> NamedThread
sanityCheckerDailyThread urlrenderer = namedThread "SanityCheckerDaily" $ forever $ do
	waitForNextCheck

	debug ["starting sanity check"]
	void $ alertWhile sanityCheckAlert go
	debug ["sanity check complete"]
  where
	go = do
		modifyDaemonStatus_ $ \s -> s { sanityCheckRunning = True }

		now <- liftIO getPOSIXTime -- before check started
		r <- either showerr return
			=<< (tryIO . batch) <~> dailyCheck urlrenderer

		modifyDaemonStatus_ $ \s -> s
			{ sanityCheckRunning = False
			, lastSanityCheck = Just now
			}

		return r

	showerr e = do
		liftAnnex $ warning $ show e
		return False

{- Only run one check per day, from the time of the last check. -}
waitForNextCheck :: Assistant ()
waitForNextCheck = do
	v <- lastSanityCheck <$> getDaemonStatus
	now <- liftIO getPOSIXTime
	liftIO $ threadDelaySeconds $ Seconds $ calcdelay now v
  where
	calcdelay _ Nothing = oneDay
	calcdelay now (Just lastcheck)
		| lastcheck < now = max oneDay $
			oneDay - truncate (now - lastcheck)
		| otherwise = oneDay

{- It's important to stay out of the Annex monad as much as possible while
 - running potentially expensive parts of this check, since remaining in it
 - will block the watcher. -}
dailyCheck :: UrlRenderer -> Assistant Bool
dailyCheck urlrenderer = do
	g <- liftAnnex gitRepo
	batchmaker <- liftIO getBatchCommandMaker

	-- Find old unstaged symlinks, and add them to git.
	(unstaged, cleanup) <- liftIO $ Git.LsFiles.notInRepo False ["."] g
	now <- liftIO getPOSIXTime
	forM_ unstaged $ \file -> do
		ms <- liftIO $ catchMaybeIO $ getSymbolicLinkStatus file
		case ms of
			Just s	| toonew (statusChangeTime s) now -> noop
				| isSymbolicLink s -> addsymlink file ms
			_ -> noop
	liftIO $ void cleanup

	{- Allow git-gc to run once per day. More frequent gc is avoided
	 - by default to avoid slowing things down. Only run repacks when 100x
	 - the usual number of loose objects are present; we tend
	 - to have a lot of small objects and they should not be a
	 - significant size. -}
	when (Git.Config.getMaybe "gc.auto" g == Just "0") $
		liftIO $ void $ Git.Command.Batch.run batchmaker
			[ Param "-c", Param "gc.auto=670000"
			, Param "gc"
			, Param "--auto"
			] g

	{- Check if the unused files found last time have been dealt with. -}
	checkOldUnused urlrenderer

	{- Run git-annex unused once per day. This is run as a separate
	 - process to stay out of the annex monad and so it can run as a
	 - batch job. -}
	program <- liftIO readProgramFile
	let (program', params') = batchmaker (program, [Param "unused"])
	void $ liftIO $ boolSystem program' params'
	{- Invalidate unused keys cache, and queue transfers of all unused
	 - keys, or if no transfers are called for, drop them. -}
	unused <- liftAnnex unusedKeys'
	void $ liftAnnex $ setUnusedKeys unused
	forM_ unused $ \k -> do
		unlessM (queueTransfers "unused" Later k Nothing Upload) $
			handleDrops "unused" True k Nothing Nothing

	return True
  where
	toonew timestamp now = now < (realToFrac (timestamp + slop) :: POSIXTime)
	slop = fromIntegral tenMinutes
	insanity msg = do
		liftAnnex $ warning msg
		void $ addAlert $ sanityCheckFixAlert msg
	addsymlink file s = do
		isdirect <- liftAnnex isDirect
		Watcher.runHandler (Watcher.onAddSymlink isdirect) file s
		insanity $ "found unstaged symlink: " ++ file

hourlyCheck :: Assistant ()
hourlyCheck = do
#ifndef mingw32_HOST_OS
	checkLogSize 0
#else
	noop
#endif

#ifndef mingw32_HOST_OS
{- Rotate logs once when total log file size is > 2 mb.
 -
 - If total log size is larger than the amount of free disk space,
 - continue rotating logs until size is < 2 mb, even if this
 - results in immediately losing the just logged data.
 -}
checkLogSize :: Int -> Assistant ()
checkLogSize n = do
	f <- liftAnnex $ fromRepo gitAnnexLogFile
	logs <- liftIO $ listLogs f
	totalsize <- liftIO $ sum <$> mapM filesize logs
	when (totalsize > 2 * oneMegabyte) $ do
		notice ["Rotated logs due to size:", show totalsize]
		liftIO $ openLog f >>= handleToFd >>= redirLog
		when (n < maxLogs + 1) $ do
			df <- liftIO $ getDiskFree $ takeDirectory f
			case df of
				Just free
					| free < fromIntegral totalsize ->
						checkLogSize (n + 1)
				_ -> noop
  where
	filesize f = fromIntegral . fileSize <$> liftIO (getFileStatus f)

	oneMegabyte :: Int
	oneMegabyte = 1000000
#endif

oneHour :: Int
oneHour = 60 * 60

oneDay :: Int
oneDay = 24 * oneHour

{- If annex.expireunused is set, find any keys that have lingered unused
 - for the specified duration, and remove them.
 -
 - Otherwise, check to see if unused keys are piling up, and let the user
 - know. -}
checkOldUnused :: UrlRenderer -> Assistant ()
checkOldUnused urlrenderer = go =<< annexExpireUnused <$> liftAnnex Annex.getGitConfig
  where
	go (Just Nothing) = noop
  	go (Just (Just expireunused)) = expireUnused (Just expireunused)
	go Nothing = maybe noop prompt =<< describeUnusedWhenBig

	prompt msg =
#ifdef WITH_WEBAPP
		do
			button <- mkAlertButton True (T.pack "Configure") urlrenderer ConfigUnusedR
			void $ addAlert $ unusedFilesAlert [button] $ T.unpack $ renderTense Present msg
#else
		debug [show $ renderTense Past msg]
#endif

{- Files may be left in misctmp by eg, an interrupted add of files
 - by the assistant, which hard links files to there as part of lockdown
 - checks. Delete these files if they're more than a day old.
 -
 - Note that this is not safe to run after the Watcher starts up, since it
 - will create such files, and due to hard linking they may have old
 - mtimes. So, this should only be called from the
 - sanityCheckerStartupThread, which runs before the Watcher starts up.
 -
 - Also, if a git-annex add is being run at the same time the assistant
 - starts up, its tmp files could be deleted. However, the watcher will
 - come along and add everything once it starts up anyway, so at worst
 - this would make the git-annex add fail unexpectedly.
 -}
cleanOldTmpMisc :: Annex ()
cleanOldTmpMisc = do
	now <- liftIO getPOSIXTime
	let oldenough = now - (60 * 60 * 24)
	tmp <- fromRepo gitAnnexTmpMiscDir
	liftIO $ mapM_ (cleanOld (<= oldenough)) =<< dirContentsRecursive tmp

{- While .git/annex/tmp is now only used for storing partially transferred
 - objects, older versions of git-annex used it for misctemp. Clean up any
 - files that might be left from that, by looking for files whose names
 - cannot be the key of an annexed object. Only delete files older than
 - 1 week old.
 -
 - Also, some remotes such as rsync may use this temp directory for storing
 - eg, encrypted objects that are being transferred. So, delete old
 - objects that use a GPGHMAC backend.
 -}
cleanReallyOldTmp :: Annex ()
cleanReallyOldTmp = do
	now <- liftIO getPOSIXTime
	let oldenough = now - (60 * 60 * 24 * 7)
	tmp <- fromRepo gitAnnexTmpObjectDir
	liftIO $ mapM_ (cleanjunk (<= oldenough)) =<< dirContentsRecursive tmp
  where
	cleanjunk check f = case fileKey (takeFileName f) of
		Nothing -> cleanOld check f
		Just k
			| "GPGHMAC" `isPrefixOf` keyBackendName k ->
				cleanOld check f
			| otherwise -> noop

cleanOld :: (POSIXTime -> Bool) -> FilePath -> IO ()
cleanOld check f = go =<< catchMaybeIO getmtime
  where
	getmtime = realToFrac . modificationTime <$> getSymbolicLinkStatus f
	go (Just mtime) | check mtime = nukeFile f
	go _ = noop


code:
{- git-annex assistant transfer polling thread
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.TransferPoller where

import Assistant.Common
import Assistant.DaemonStatus
import Logs.Transfer
import Utility.NotificationBroadcaster
import qualified Assistant.Threads.TransferWatcher as TransferWatcher

import Control.Concurrent
import qualified Data.Map as M

{- This thread polls the status of ongoing transfers, determining how much
 - of each transfer is complete. -}
transferPollerThread :: NamedThread
transferPollerThread = namedThread "TransferPoller" $ do
	g <- liftAnnex gitRepo
	tn <- liftIO . newNotificationHandle True =<<
		transferNotifier <$> getDaemonStatus
	forever $ do
		liftIO $ threadDelay 500000 -- 0.5 seconds
		ts <- currentTransfers <$> getDaemonStatus
		if M.null ts
			-- block until transfers running
			then liftIO $ waitNotification tn
			else mapM_ (poll g) $ M.toList ts
  where
	poll g (t, info)
		{- Downloads are polled by checking the size of the
		 - temp file being used for the transfer. -}
		| transferDirection t == Download = do
			let f = gitAnnexTmpObjectLocation (transferKey t) g
			sz <- liftIO $ catchMaybeIO $
				fromIntegral . fileSize <$> getFileStatus f
			newsize t info sz
		{- Uploads don't need to be polled for when the TransferWatcher
		 - thread can track file modifications. -}
		| TransferWatcher.watchesTransferSize = noop
		{- Otherwise, this code polls the upload progress
		 - by reading the transfer info file. -}
		| otherwise = do
			let f = transferFile t g
			mi <- liftIO $ catchDefaultIO Nothing $
				readTransferInfoFile Nothing f
			maybe noop (newsize t info . bytesComplete) mi

	newsize t info sz
		| bytesComplete info /= sz && isJust sz =
			alterTransferInfo t $ \i -> i { bytesComplete = sz }
		| otherwise = noop


code:
{- git-annex assistant thread to scan remotes to find needed transfers
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.TransferScanner where

import Assistant.Common
import Assistant.Types.ScanRemotes
import Assistant.ScanRemotes
import Assistant.TransferQueue
import Assistant.DaemonStatus
import Assistant.Drop
import Assistant.Sync
import Assistant.DeleteRemote
import Assistant.Types.UrlRenderer
import Logs.Transfer
import Logs.Location
import Logs.Group
import Logs.Web (webUUID)
import qualified Remote
import qualified Types.Remote as Remote
import Utility.ThreadScheduler
import Utility.NotificationBroadcaster
import Utility.Batch
import qualified Git.LsFiles as LsFiles
import qualified Backend
import Annex.Content
import Annex.Wanted
import CmdLine.Action

import qualified Data.Set as S

{- This thread waits until a remote needs to be scanned, to find transfers
 - that need to be made, to keep data in sync.
 -}
transferScannerThread :: UrlRenderer -> NamedThread
transferScannerThread urlrenderer = namedThread "TransferScanner" $ do
	startupScan
	go S.empty
  where
	go scanned = do
		scanrunning False
		liftIO $ threadDelaySeconds (Seconds 2)
		(rs, infos) <- unzip <$> getScanRemote
		scanrunning True
		if any fullScan infos || any (`S.notMember` scanned) rs
			then do
				expensiveScan urlrenderer rs
				go $ scanned `S.union` S.fromList rs
			else do
				mapM_ failedTransferScan rs
				go scanned
	scanrunning b = do
		ds <- modifyDaemonStatus $ \s ->
			(s { transferScanRunning = b }, s)
		liftIO $ sendNotification $ transferNotifier ds

	{- All git remotes are synced, and all available remotes
	 - are scanned in full on startup, for multiple reasons, including:
	 -
	 - * This may be the first run, and there may be remotes
	 -   already in place, that need to be synced.
	 - * Changes may have been made last time we run, but remotes were
	 -   not available to be synced with.
	 - * Changes may have been made to remotes while we were down.
	 - * We may have run before, and scanned a remote, but
	 -   only been in a subdirectory of the git remote, and so
	 -   not synced it all.
	 - * We may have run before, and had transfers queued,
	 -   and then the system (or us) crashed, and that info was
	 -   lost.
	 - * A remote may be in the unwanted group, and this is a chance
	 -   to determine if the remote has been emptied.
	 -}
	startupScan = do
		reconnectRemotes True =<< syncGitRemotes <$> getDaemonStatus
		addScanRemotes True =<< syncDataRemotes <$> getDaemonStatus

{- This is a cheap scan for failed transfers involving a remote. -}
failedTransferScan :: Remote -> Assistant ()
failedTransferScan r = do
	failed <- liftAnnex $ clearFailedTransfers (Remote.uuid r)
	mapM_ retry failed
  where
	retry (t, info)
		| transferDirection t == Download =
			{- Check if the remote still has the key.
			 - If not, relies on the expensiveScan to
			 - get it queued from some other remote. -}
			whenM (liftAnnex $ remoteHas r $ transferKey t) $
				requeue t info
		| otherwise =
			{- The Transferrer checks when uploading
			 - that the remote doesn't already have the
			 - key, so it's not redundantly checked here. -}
			requeue t info
	requeue t info = queueTransferWhenSmall "retrying failed transfer" (associatedFile info) t r

{- This is a expensive scan through the full git work tree, finding
 - files to transfer. The scan is blocked when the transfer queue gets
 - too large.
 -
 - This also finds files that are present either here or on a remote
 - but that are not preferred content, and drops them. Searching for files
 - to drop is done concurrently with the scan for transfers.
 -
 - TODO: It would be better to first drop as much as we can, before
 - transferring much, to minimise disk use.
 -
 - During the scan, we'll also check if any unwanted repositories are empty,
 - and can be removed. While unrelated, this is a cheap place to do it,
 - since we need to look at the locations of all keys anyway.
 -}
expensiveScan :: UrlRenderer -> [Remote] -> Assistant ()
expensiveScan urlrenderer rs = unless onlyweb $ batch <~> do
	debug ["starting scan of", show visiblers]

	let us = map Remote.uuid rs

	mapM_ (liftAnnex . clearFailedTransfers) us

	unwantedrs <- liftAnnex $ S.fromList
		<$> filterM inUnwantedGroup us

	g <- liftAnnex gitRepo
	(files, cleanup) <- liftIO $ LsFiles.inRepo [] g
	removablers <- scan unwantedrs files
	void $ liftIO cleanup

	debug ["finished scan of", show visiblers]

	remove <- asIO1 $ removableRemote urlrenderer
	liftIO $ mapM_ (void . tryNonAsync . remove) $ S.toList removablers
  where
	onlyweb = all (== webUUID) $ map Remote.uuid rs
	visiblers = let rs' = filter (not . Remote.readonly) rs
		in if null rs' then rs else rs'

	scan unwanted [] = return unwanted
	scan unwanted (f:fs) = do
		(unwanted', ts) <- maybe
			(return (unwanted, []))
			(findtransfers f unwanted)
				=<< liftAnnex (Backend.lookupFile f)
		mapM_ (enqueue f) ts
		scan unwanted' fs

	enqueue f (r, t) =
		queueTransferWhenSmall "expensive scan found missing object"
			(Just f) t r
	findtransfers f unwanted key = do
		{- The syncable remotes may have changed since this
		 - scan began. -}
		syncrs <- syncDataRemotes <$> getDaemonStatus
		locs <- liftAnnex $ loggedLocations key
		present <- liftAnnex $ inAnnex key
		liftAnnex $ handleDropsFrom locs syncrs
			"expensive scan found too many copies of object"
			present key (Just f) Nothing callCommandAction
		liftAnnex $ do
			let slocs = S.fromList locs
			let use a = return $ mapMaybe (a key slocs) syncrs
			ts <- if present
				then filterM (wantSend True (Just key) (Just f) . Remote.uuid . fst)
					=<< use (genTransfer Upload False)
				else ifM (wantGet True (Just key) (Just f))
					( use (genTransfer Download True) , return [] )
			let unwanted' = S.difference unwanted slocs
			return (unwanted', ts)

genTransfer :: Direction -> Bool -> Key -> S.Set UUID -> Remote -> Maybe (Remote, Transfer)
genTransfer direction want key slocs r
	| direction == Upload && Remote.readonly r = Nothing
	| S.member (Remote.uuid r) slocs == want = Just
		(r, Transfer direction (Remote.uuid r) key)
	| otherwise = Nothing

remoteHas :: Remote -> Key -> Annex Bool
remoteHas r key = elem
	<$> pure (Remote.uuid r)
	<*> loggedLocations key


code:
{- git-annex assistant transfer watching thread
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.TransferWatcher where

import Assistant.Common
import Assistant.DaemonStatus
import Assistant.TransferSlots
import Logs.Transfer
import Utility.DirWatcher
import Utility.DirWatcher.Types
import qualified Remote

import Control.Concurrent
import qualified Data.Map as M

{- This thread watches for changes to the gitAnnexTransferDir,
 - and updates the DaemonStatus's map of ongoing transfers. -}
transferWatcherThread :: NamedThread
transferWatcherThread = namedThread "TransferWatcher" $ do
	dir <- liftAnnex $ gitAnnexTransferDir <$> gitRepo
	liftIO $ createDirectoryIfMissing True dir
	let hook a = Just <$> asIO2 (runHandler a)
	addhook <- hook onAdd
	delhook <- hook onDel
	modifyhook <- hook onModify
	errhook <- hook onErr
	let hooks = mkWatchHooks
		{ addHook = addhook
		, delHook = delhook
		, modifyHook = modifyhook
		, errHook = errhook
		}
	void $ liftIO $ watchDir dir (const False) True hooks id
	debug ["watching for transfers"]

type Handler = FilePath -> Assistant ()

{- Runs an action handler.
 -
 - Exceptions are ignored, otherwise a whole thread could be crashed.
 -}
runHandler :: Handler -> FilePath -> Maybe FileStatus -> Assistant ()
runHandler handler file _filestatus =
	either (liftIO . print) (const noop) =<< tryIO <~> handler file

{- Called when there's an error with inotify. -}
onErr :: Handler
onErr = error

{- Called when a new transfer information file is written. -}
onAdd :: Handler
onAdd file = case parseTransferFile file of
	Nothing -> noop
	Just t -> go t =<< liftAnnex (checkTransfer t)
  where
	go _ Nothing = noop -- transfer already finished
	go t (Just info) = do
		debug [ "transfer starting:", describeTransfer t info ]
		r <- liftAnnex $ Remote.remoteFromUUID $ transferUUID t
		updateTransferInfo t info { transferRemote = r }

{- Called when a transfer information file is updated.
 -
 - The only thing that should change in the transfer info is the
 - bytesComplete, so that's the only thing updated in the DaemonStatus. -}
onModify :: Handler
onModify file = case parseTransferFile file of
	Nothing -> noop
	Just t -> go t =<< liftIO (readTransferInfoFile Nothing file)
  where
	go _ Nothing = noop
	go t (Just newinfo) = alterTransferInfo t $
		\i -> i { bytesComplete = bytesComplete newinfo }

{- This thread can only watch transfer sizes when the DirWatcher supports
 - tracking modificatons to files. -}
watchesTransferSize :: Bool
watchesTransferSize = modifyTracked

{- Called when a transfer information file is removed. -}
onDel :: Handler
onDel file = case parseTransferFile file of
	Nothing -> noop
	Just t -> do
		debug [ "transfer finishing:", show t]
		minfo <- removeTransfer t

		-- Run transfer hook.
		m <- transferHook <$> getDaemonStatus
		maybe noop (\hook -> void $ liftIO $ forkIO $ hook t)
			(M.lookup (transferKey t) m)

		finished <- asIO2 finishedTransfer
		void $ liftIO $ forkIO $ do
			{- XXX race workaround delay. The location
			 - log needs to be updated before finishedTransfer
			 - runs. -}
			threadDelay 10000000 -- 10 seconds
			finished t minfo


code:
{- git-annex assistant data transferrer thread
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.Threads.Transferrer where

import Assistant.Common
import Assistant.TransferQueue
import Assistant.TransferSlots
import Logs.Transfer
import Config.Files
import Utility.Batch

{- Dispatches transfers from the queue. -}
transfererThread :: NamedThread
transfererThread = namedThread "Transferrer" $ do
	program <- liftIO readProgramFile
	batchmaker <- liftIO getBatchCommandMaker
	forever $ inTransferSlot program batchmaker $
		maybe (return Nothing) (uncurry genTransfer)
			=<< getNextTransfer notrunning
  where
	{- Skip transfers that are already running. -}
	notrunning = isNothing . startedTime


code:
{- git-annex assistant thread to detect when git-annex is upgraded
 -
 - Copyright 2013 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

{-# LANGUAGE CPP #-}

module Assistant.Threads.UpgradeWatcher (
	upgradeWatcherThread
) where

import Assistant.Common
import Assistant.Upgrade
import Utility.DirWatcher
import Utility.DirWatcher.Types
import Utility.ThreadScheduler
import Assistant.Types.UrlRenderer
import Assistant.Alert
import Assistant.DaemonStatus
#ifdef WITH_WEBAPP
import Assistant.WebApp.Types
import qualified Build.SysConfig
#endif

import Control.Concurrent.MVar
import qualified Data.Text as T

data WatcherState = InStartupScan | Started | Upgrading
	deriving (Eq)

upgradeWatcherThread :: UrlRenderer -> NamedThread
upgradeWatcherThread urlrenderer = namedThread "UpgradeWatcher" $ do
	whenM (liftIO checkSuccessfulUpgrade) $
		showSuccessfulUpgrade urlrenderer
	go =<< liftIO upgradeFlagFile
  where
	go Nothing = debug [ "cannot determine program path" ]
	go (Just flagfile) = do
		mvar <- liftIO $ newMVar InStartupScan
		changed <- Just <$> asIO2 (changedFile urlrenderer mvar flagfile)
		let hooks = mkWatchHooks
			{ addHook = changed
			, delHook = changed
			, addSymlinkHook = changed
			, modifyHook = changed
			, delDirHook = changed
			}
		let dir = parentDir flagfile
		let depth = length (splitPath dir) + 1
		let nosubdirs f = length (splitPath f) == depth
		void $ liftIO $ watchDir dir nosubdirs False hooks (startup mvar)
  	-- Ignore bogus events generated during the startup scan.
	-- We ask the watcher to not generate them, but just to be safe..
  	startup mvar scanner = do
		r <- scanner
		void $ swapMVar mvar Started
		return r

changedFile :: UrlRenderer -> MVar WatcherState -> FilePath -> FilePath -> Maybe FileStatus -> Assistant ()
changedFile urlrenderer mvar flagfile file _status
	| flagfile /= file = noop
	| otherwise = do
		state <- liftIO $ readMVar mvar
		when (state == Started) $ do
			setstate Upgrading
			ifM (liftIO upgradeSanityCheck)
				( handleUpgrade urlrenderer
				, do
					debug ["new version failed sanity check; not using"]
					setstate Started
				)
  where
	setstate = void . liftIO . swapMVar mvar

handleUpgrade :: UrlRenderer -> Assistant ()
handleUpgrade urlrenderer = do
	-- Wait 2 minutes for any final upgrade changes to settle.
	-- (For example, other associated files may be being put into
	-- place.) Not needed when using a distribution bundle, because
	-- in that case git-annex handles the upgrade in a non-racy way.
	liftIO $ unlessM usingDistribution $
		threadDelaySeconds (Seconds 120)
	ifM autoUpgradeEnabled
		( do
			debug ["starting automatic upgrade"]
			unattendedUpgrade
#ifdef WITH_WEBAPP
		, do
			button <- mkAlertButton True (T.pack "Finish Upgrade") urlrenderer ConfigFinishUpgradeR
			void $ addAlert $ upgradeReadyAlert button
#else
		, noop
#endif
		)

showSuccessfulUpgrade :: UrlRenderer -> Assistant ()
showSuccessfulUpgrade urlrenderer = do
#ifdef WITH_WEBAPP
	button <- ifM autoUpgradeEnabled
		( pure Nothing
		, Just <$> mkAlertButton True
			(T.pack "Enable Automatic Upgrades")
			urlrenderer ConfigEnableAutomaticUpgradeR
		)
	void $ addAlert $ upgradeFinishedAlert button Build.SysConfig.packageversion
#else
	noop
#endif


code:
{- git-annex assistant thread to detect when upgrade is available
 -
 - Copyright 2013 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

{-# LANGUAGE CPP #-}

module Assistant.Threads.Upgrader (
	upgraderThread
) where

import Assistant.Common
import Assistant.Upgrade

import Assistant.Types.UrlRenderer
import Assistant.DaemonStatus
import Assistant.Alert
import Utility.NotificationBroadcaster
import qualified Annex
import qualified Build.SysConfig
import qualified Git.Version
import Types.Distribution
#ifdef WITH_WEBAPP
import Assistant.WebApp.Types
#endif

import Data.Time.Clock
import qualified Data.Text as T

upgraderThread :: UrlRenderer -> NamedThread
upgraderThread urlrenderer = namedThread "Upgrader" $
	when (isJust Build.SysConfig.upgradelocation) $ do
		{- Check for upgrade on startup, unless it was just
		 - upgraded. -}
		unlessM (liftIO checkSuccessfulUpgrade) $
			checkUpgrade urlrenderer
		h <- liftIO . newNotificationHandle False . networkConnectedNotifier =<< getDaemonStatus
		go h =<< liftIO getCurrentTime
  where
  	{- Wait for a network connection event. Then see if it's been
	 - half a day since the last upgrade check. If so, proceed with
	 - check. -}
	go h lastchecked = do
		liftIO $ waitNotification h
		autoupgrade <- liftAnnex $ annexAutoUpgrade <$> Annex.getGitConfig
		if autoupgrade == NoAutoUpgrade
			then go h lastchecked
			else do
				now <- liftIO getCurrentTime
				if diffUTCTime now lastchecked > halfday
					then do
						checkUpgrade urlrenderer
						go h =<< liftIO getCurrentTime
					else go h lastchecked
	halfday = 12 * 60 * 60

checkUpgrade :: UrlRenderer -> Assistant ()
checkUpgrade urlrenderer = do
	debug [ "Checking if an upgrade is available." ]
	go =<< downloadDistributionInfo
  where
	go Nothing = debug [ "Failed to check if upgrade is available." ]
	go (Just d) = do
		let installed = Git.Version.normalize Build.SysConfig.packageversion
		let avail = Git.Version.normalize $ distributionVersion d
		let old = Git.Version.normalize <$> distributionUrgentUpgrade d
		if Just installed <= old
			then canUpgrade High urlrenderer d
			else if installed < avail
				then canUpgrade Low urlrenderer d
				else debug [ "No new version found." ]

canUpgrade :: AlertPriority -> UrlRenderer -> GitAnnexDistribution -> Assistant ()
canUpgrade urgency urlrenderer d = ifM autoUpgradeEnabled
	( startDistributionDownload d
	, do
#ifdef WITH_WEBAPP
		button <- mkAlertButton True (T.pack "Upgrade") urlrenderer (ConfigStartUpgradeR d)
		void $ addAlert (canUpgradeAlert urgency (distributionVersion d) button)
#else
		noop
#endif
	)


code:
{- git-annex assistant tree watcher
 -
 - Copyright 2012-2013 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

{-# LANGUAGE DeriveDataTypeable, CPP #-}

module Assistant.Threads.Watcher (
	watchThread,
	WatcherControl(..),
	checkCanWatch,
	needLsof,
	onAddSymlink,
	runHandler,
) where

import Assistant.Common
import Assistant.DaemonStatus
import Assistant.Changes
import Assistant.Types.Changes
import Assistant.Alert
import Utility.DirWatcher
import Utility.DirWatcher.Types
import qualified Annex
import qualified Annex.Queue
import qualified Git
import qualified Git.UpdateIndex
import qualified Git.LsFiles as LsFiles
import qualified Backend
import Annex.Direct
import Annex.Content.Direct
import Annex.CatFile
import Annex.CheckIgnore
import Annex.Link
import Annex.FileMatcher
import Types.FileMatcher
import Annex.ReplaceFile
import Git.Types
import Config
import Utility.ThreadScheduler
#ifndef mingw32_HOST_OS
import qualified Utility.Lsof as Lsof
#endif

import Data.Bits.Utils
import Data.Typeable
import qualified Data.ByteString.Lazy as L
import qualified Control.Exception as E
import Data.Time.Clock

checkCanWatch :: Annex ()
checkCanWatch
	| canWatch = do
#ifndef mingw32_HOST_OS
		liftIO Lsof.setup
		unlessM (liftIO (inPath "lsof") <||> Annex.getState Annex.force)
			needLsof
#else
		noop
#endif
	| otherwise = error "watch mode is not available on this system"

needLsof :: Annex ()
needLsof = error $ unlines
	[ "The lsof command is needed for watch mode to be safe, and is not in PATH."
	, "To override lsof checks to ensure that files are not open for writing"
	, "when added to the annex, you can use --force"
	, "Be warned: This can corrupt data in the annex, and make fsck complain."
	]

{- A special exception that can be thrown to pause or resume the watcher. -}
data WatcherControl = PauseWatcher | ResumeWatcher
        deriving (Show, Eq, Typeable)

instance E.Exception WatcherControl

watchThread :: NamedThread
watchThread = namedThread "Watcher" $
	ifM (liftAnnex $ annexAutoCommit <$> Annex.getGitConfig)
		( runWatcher
		, waitFor ResumeWatcher runWatcher
		)

runWatcher :: Assistant ()
runWatcher = do
	startup <- asIO1 startupScan
	matcher <- liftAnnex largeFilesMatcher
	direct <- liftAnnex isDirect
	symlinkssupported <- liftAnnex $ coreSymlinks <$> Annex.getGitConfig
	addhook <- hook $ if direct
		then onAddDirect symlinkssupported matcher
		else onAdd matcher
	delhook <- hook onDel
	addsymlinkhook <- hook $ onAddSymlink direct
	deldirhook <- hook onDelDir
	errhook <- hook onErr
	let hooks = mkWatchHooks
		{ addHook = addhook
		, delHook = delhook
		, addSymlinkHook = addsymlinkhook
		, delDirHook = deldirhook
		, errHook = errhook
		}
	scanevents <- liftAnnex $ annexStartupScan <$> Annex.getGitConfig
	handle <- liftIO $ watchDir "." ignored scanevents hooks startup
	debug [ "watching", "."]

	{- Let the DirWatcher thread run until signalled to pause it,
	 - then wait for a resume signal, and restart. -}
	waitFor PauseWatcher $ do
		liftIO $ stopWatchDir handle
		waitFor ResumeWatcher runWatcher
  where
	hook a = Just <$> asIO2 (runHandler a)

waitFor :: WatcherControl -> Assistant () -> Assistant ()
waitFor sig next = do
	r <- liftIO (E.try pause :: IO (Either E.SomeException ()))
	case r of
		Left e -> case E.fromException e of
			Just s
				| s == sig -> next
			_ -> noop
		_ -> noop
  where
	pause = runEvery (Seconds 86400) noop

{- Initial scartup scan. The action should return once the scan is complete. -}
startupScan :: IO a -> Assistant a
startupScan scanner = do
	liftAnnex $ showAction "scanning"
	alertWhile' startupScanAlert $ do
		r <- liftIO scanner

		-- Notice any files that were deleted before
		-- watching was started.
		top <- liftAnnex $ fromRepo Git.repoPath
		(fs, cleanup) <- liftAnnex $ inRepo $ LsFiles.deleted [top]
		forM_ fs $ \f -> do
			liftAnnex $ onDel' f
			maybe noop recordChange =<< madeChange f RmChange
		void $ liftIO cleanup

		liftAnnex $ showAction "started"
		liftIO $ putStrLn ""

		modifyDaemonStatus_ $ \s -> s { scanComplete = True }

		-- Ensure that the Committer sees any changes
		-- that it did not process, and acts on them now that
		-- the scan is complete.
		refillChanges =<< getAnyChanges

		return (True, r)

{- Hardcoded ignores, passed to the DirWatcher so it can avoid looking
 - at the entire .git directory. Does not include .gitignores. -}
ignored :: FilePath -> Bool
ignored = ig . takeFileName
  where
	ig ".git" = True
	ig ".gitignore" = True
	ig ".gitattributes" = True
#ifdef darwin_HOST_OS
	ig ".DS_Store" = True
#endif
	ig _ = False

unlessIgnored :: FilePath -> Assistant (Maybe Change) -> Assistant (Maybe Change)
unlessIgnored file a = ifM (liftAnnex $ checkIgnored file)
	( noChange
	, a
	)

type Handler = FilePath -> Maybe FileStatus -> Assistant (Maybe Change)

{- Runs an action handler, and if there was a change, adds it to the ChangeChan.
 -
 - Exceptions are ignored, otherwise a whole watcher thread could be crashed.
 -}
runHandler :: Handler -> FilePath -> Maybe FileStatus -> Assistant ()
runHandler handler file filestatus = void $ do
	r <- tryIO <~> handler (normalize file) filestatus
	case r of
		Left e -> liftIO $ warningIO $ show e
		Right Nothing -> noop
		Right (Just change) -> do
			-- Just in case the commit thread is not
			-- flushing the queue fast enough.
			liftAnnex Annex.Queue.flushWhenFull
			recordChange change
  where
  	normalize f
		| "./" `isPrefixOf` file = drop 2 f
		| otherwise = f

{- Small files are added to git as-is, while large ones go into the annex. -}
add :: FileMatcher Annex -> FilePath -> Assistant (Maybe Change)
add bigfilematcher file = ifM (liftAnnex $ checkFileMatcher bigfilematcher file)
	( pendingAddChange file
	, do
		liftAnnex $ Annex.Queue.addCommand "add"
			[Params "--force --"] [file]
		madeChange file AddFileChange
	)

onAdd :: FileMatcher Annex -> Handler
onAdd matcher file filestatus
	| maybe False isRegularFile filestatus =
		unlessIgnored file $
			add matcher file
	| otherwise = noChange

shouldRestage :: DaemonStatus -> Bool
shouldRestage ds = scanComplete ds || forceRestage ds

{- In direct mode, add events are received for both new files, and
 - modified existing files.
 -}
onAddDirect :: Bool -> FileMatcher Annex -> Handler
onAddDirect symlinkssupported matcher file fs = do
	v <- liftAnnex $ catKeyFile file
	case (v, fs) of
		(Just key, Just filestatus) ->
			ifM (liftAnnex $ sameFileStatus key filestatus)
				{- It's possible to get an add event for
				 - an existing file that is not
				 - really modified, but it might have
				 - just been deleted and been put back,
				 - so it symlink is restaged to make sure. -}
				( ifM (shouldRestage <$> getDaemonStatus)
					( do
						link <- liftAnnex $ inRepo $ gitAnnexLink file key
						addLink file link (Just key)
					, noChange
					)
				, guardSymlinkStandin (Just key) $ do
					debug ["changed direct", file]
					liftAnnex $ changedDirect key file
					add matcher file
				)
		_ -> unlessIgnored file $
			guardSymlinkStandin Nothing $ do
				debug ["add direct", file]
				add matcher file
  where
 	{- On a filesystem without symlinks, we'll get changes for regular
	 - files that git uses to stand-in for symlinks. Detect when
	 - this happens, and stage the symlink, rather than annexing the
	 - file. -}
	guardSymlinkStandin mk a
		| symlinkssupported = a
		| otherwise = do
			linktarget <- liftAnnex $ getAnnexLinkTarget file
			case linktarget of
				Nothing -> a
				Just lt -> do
					case fileKey $ takeFileName lt of
						Nothing -> noop
						Just key -> void $ liftAnnex $
							addAssociatedFile key file
					onAddSymlink' linktarget mk True file fs

{- A symlink might be an arbitrary symlink, which is just added.
 - Or, if it is a git-annex symlink, ensure it points to the content
 - before adding it.
 -}
onAddSymlink :: Bool -> Handler
onAddSymlink isdirect file filestatus = unlessIgnored file $ do
	linktarget <- liftIO (catchMaybeIO $ readSymbolicLink file)
	kv <- liftAnnex (Backend.lookupFile file)
	onAddSymlink' linktarget kv isdirect file filestatus

onAddSymlink' :: Maybe String -> Maybe Key -> Bool -> Handler
onAddSymlink' linktarget mk isdirect file filestatus = go mk
  where
  	go (Just key) = do
		when isdirect $
			liftAnnex $ void $ addAssociatedFile key file
		link <- liftAnnex $ inRepo $ gitAnnexLink file key
		if linktarget == Just link
			then ensurestaged (Just link) =<< getDaemonStatus
			else do
				unless isdirect $
					liftAnnex $ replaceFile file $
						makeAnnexLink link
				addLink file link (Just key)
	-- other symlink, not git-annex
	go Nothing = ensurestaged linktarget =<< getDaemonStatus

	{- This is often called on symlinks that are already
	 - staged correctly. A symlink may have been deleted
	 - and being re-added, or added when the watcher was
	 - not running. So they're normally restaged to make sure.
	 -
	 - As an optimisation, during the startup scan, avoid
	 - restaging everything. Only links that were created since
	 - the last time the daemon was running are staged.
	 - (If the daemon has never ran before, avoid staging
	 - links too.)
	 -}
	ensurestaged (Just link) daemonstatus
		| shouldRestage daemonstatus = addLink file link mk
		| otherwise = case filestatus of
			Just s
				| not (afterLastDaemonRun (statusChangeTime s) daemonstatus) -> noChange
			_ -> addLink file link mk
	ensurestaged Nothing _ = noChange

{- For speed, tries to reuse the existing blob for symlink target. -}
addLink :: FilePath -> FilePath -> Maybe Key -> Assistant (Maybe Change)
addLink file link mk = do
	debug ["add symlink", file]
	liftAnnex $ do
		v <- catObjectDetails $ Ref $ ':':file
		case v of
			Just (currlink, sha, _type)
				| s2w8 link == L.unpack currlink ->
					stageSymlink file sha
			_ -> stageSymlink file =<< hashSymlink link
	madeChange file $ LinkChange mk

onDel :: Handler
onDel file _ = do
	debug ["file deleted", file]
	liftAnnex $ onDel' file
	madeChange file RmChange

onDel' :: FilePath -> Annex ()
onDel' file = do
	whenM isDirect $ do
		mkey <- catKeyFile file
		case mkey of
			Nothing -> noop
			Just key -> void $ removeAssociatedFile key file
	Annex.Queue.addUpdateIndex =<<
		inRepo (Git.UpdateIndex.unstageFile file)

{- A directory has been deleted, or moved, so tell git to remove anything
 - that was inside it from its cache. Since it could reappear at any time,
 - use --cached to only delete it from the index.
 -
 - This queues up a lot of RmChanges, which assists the Committer in
 - pairing up renamed files when the directory was renamed. -}
onDelDir :: Handler
onDelDir dir _ = do
	debug ["directory deleted", dir]
	(fs, clean) <- liftAnnex $ inRepo $ LsFiles.deleted [dir]

	liftAnnex $ mapM_ onDel' fs

	-- Get the events queued up as fast as possible, so the
	-- committer sees them all in one block.
	now <- liftIO getCurrentTime
	recordChanges $ map (\f -> Change now f RmChange) fs

	void $ liftIO clean
	liftAnnex Annex.Queue.flushWhenFull
	noChange

{- Called when there's an error with inotify or kqueue. -}
onErr :: Handler
onErr msg _ = do
	liftAnnex $ warning msg
	void $ addAlert $ warningAlert "watcher" msg
	noChange